//! Authoritative Stage 3c2 v4 run-save and replay wire protocols.
//!
//! The top-level envelopes retain immutable run identity, ordered lifecycle
//! events, and (for Story runs) one nonrecursive town checkpoint. The explicit
//! wire records below mirror core plain-data types without requiring persistence
//! derives in `hexwork-core`. Every enum is an explicit numeric tag and every
//! `f32` is stored as its exact IEEE-754 bits.

use std::collections::BTreeMap;
use std::fmt;
use std::io::{Read, Write};
use std::path::Path;

use crate::knowledge::{
    validate_initial_lineage, AuctionLot, GhostMemory, InitialLineage, LineageId, LotId, RunId,
    MAX_AUCTION_LOTS, MAX_GHOSTS, MAX_GHOST_WARNING_BYTES, MAX_IDENTIFIED_TEMPLATES,
    MAX_INITIAL_ROWS, MAX_INITIAL_STRING_BYTES, MAX_INITIAL_TAGS, MAX_LOT_TAGS, MAX_LOT_TAG_BYTES,
    MAX_MAPPED_REGIONS, MAX_RECIPES, MAX_TRANSLATIONS,
};
use bincode::de::Decoder;
use bincode::error::DecodeError;
use bincode::{Decode, Encode};
use hexwork_core::action::{CommittedAction, Direction};
use hexwork_core::actor::{Actor, Stance};
use hexwork_core::bargain::{Debt, Ledger};
use hexwork_core::character::{Ancestry, Calling, Character};
use hexwork_core::checkpoint::{CauseImage, ChunkImage, LawImage, StateImage};
use hexwork_core::connector::{Connector, ConnectorKind};
use hexwork_core::dungeon::{Chalk, MulePath, VaultChoice};
use hexwork_core::hymn::{HymnCard, HymnCase};
use hexwork_core::inventory::Pack;
use hexwork_core::item::{Condition, Instance};
use hexwork_core::light::LightSource;
use hexwork_core::modes::{ChapelDebt, Mode};
use hexwork_core::npc::{Job, ShopMemory, Townsfolk};
use hexwork_core::observation::{
    Face, LocationKey, ObservationRow, ObservedFace, DOOR_APPEARANCE_GLASS, LOCATION_KIND_DEEP,
    LOCATION_KIND_TOWN, OBSERVATION_TOTAL_CAP,
};
use hexwork_core::pending::{Pending, PendingAction, PendingInn, PendingTag};
use hexwork_core::pressure::{PressureEdge, PressureNode};
use hexwork_core::reputation::{Faction, GodId, Reputation, Standing};
use hexwork_core::rot::CorpseClock;
use hexwork_core::runtime::{
    ActiveContract, ActorIdentity, ActorRole, ContractStatus, RunKnowledge, RunLocation, Wallet,
    KNOWLEDGE_ID_MAX_BYTES, KNOWLEDGE_TOTAL_ID_BYTES_LIMIT, KNOWLEDGE_TOTAL_ROW_LIMIT,
    MAPPED_REGION_LIMIT, RECIPE_LIMIT, TRANSLATION_LIMIT,
};
use hexwork_core::services::Service;
use hexwork_core::sound::SoundEvent;
use hexwork_core::state::{FarField, GossipPacket, Spill, State, CANONICAL_HASH_VERSION};
use hexwork_core::voxel::{Voxel, CHUNK_VOL};
use sha2::{Digest, Sha256};
use std::ops::{Deref, DerefMut};

/// Current authoritative run-save format. Pre-release v1-v3 artifacts have no
/// migration and are rejected explicitly after the bounded raw header is read.
pub const RUN_FORMAT_VERSION: u32 = 0x8005_0005; // development fixed schema, not final v5
/// Current production replay format. V4 stores only identity, lifecycle events,
/// and terminal tick/hash; presentation evidence is derived during replay.
pub const REPLAY_FORMAT_VERSION: u32 = 0x8005_0005; // development fixed schema, not final v5
/// Largest compressed run/replay accepted at an untrusted boundary (64 MiB).
pub const MAX_COMPRESSED_BYTES: usize = 64 * 1024 * 1024;
/// Largest decompressed run/replay accepted while streaming (256 MiB). This is
/// also the exact bincode encode/decode claim limit, so the encoder cannot emit
/// an envelope rejected by a smaller parser backstop.
pub const MAX_DECOMPRESSED_BYTES: usize = 256 * 1024 * 1024;
/// Maximum chunks in one State image.
pub const MAX_CHUNKS: usize = 64;
/// Aggregate chunks across current and optional Story checkpoint images. The
/// current generated Story shape is 54+54 chunks; 128 also admits the bounded
/// worst case of 64+64 without resetting the budget between images.
pub const MAX_TOTAL_CHUNKS: usize = 128;
/// Maximum rows in any ordinary state table.
pub const MAX_ROWS: usize = 131_072;
/// Maximum ordered lifecycle events in one run or replay.
pub const MAX_SESSION_EVENTS: usize = 65_536;
/// Maximum UTF-8 bytes in one string field.
pub const MAX_STRING_BYTES: usize = 4 * 1024;
/// Maximum cumulative string bytes in an envelope.
pub const MAX_TOTAL_STRING_BYTES: usize = 16 * 1024 * 1024;
/// Aggregate rows across all top-level `StateWire` tables.
pub const MAX_TOTAL_STATE_ROWS: usize = 262_144;
/// Aggregate nested rows across the single envelope. 16,000,000 exceeds all
/// 128 permitted complete 48³ chunks (14,155,776 voxels) while leaving a finite
/// margin for the separately capped nested category rows.
pub const MAX_TOTAL_NESTED_ROWS: usize = 16_000_000;
/// Aggregate spill gas-flow rows.
pub const MAX_TOTAL_GAS_FLOWS: usize = 262_144;
/// Aggregate mule waypoints.
pub const MAX_TOTAL_MULE_WAYPOINTS: usize = 65_536;
/// Aggregate reputation town-ledger rows.
pub const MAX_TOTAL_TOWN_LEDGER_ROWS: usize = 262_144;
/// Aggregate items carried across all packs.
pub const MAX_TOTAL_PACK_ITEMS: usize = 262_144;
/// Aggregate hymn cards and card miscast rows.
pub const MAX_TOTAL_HYMN_CARDS: usize = 65_536;
pub const MAX_TOTAL_HYMN_MISCASTS: usize = 262_144;
/// Aggregate bargain debt rows.
pub const MAX_TOTAL_LEDGER_DEBTS: usize = 262_144;
/// Aggregate shop broke/stole/killed memory rows.
pub const MAX_TOTAL_SHOP_ROWS: usize = 262_144;
/// Aggregate item tags across loose and packed items.
pub const MAX_TOTAL_ITEM_TAGS: usize = 262_144;

const MAX_ITEM_TAGS: usize = 256;
/// The single pending-action slot is at most one row (docs/fluid §4.3).
const MAX_PENDING_ROWS: usize = 1;
/// Observation-memory rows share the core's aggregate cap (docs/fluid §5).
const MAX_OBSERVATION_ROWS: usize = OBSERVATION_TOTAL_CAP;
const MAX_PACK_ITEMS: usize = 8_192;
const MAX_HYMN_CARDS: usize = 256;
const MAX_HYMN_MISCASTS: usize = 256;
const MAX_LEDGER_DEBTS: usize = 4_096;
const MAX_TOWN_LEDGER_ROWS: usize = 8_192;
const MAX_SHOP_ROWS: usize = 65_536;
const ZSTD_LEVEL: i32 = 3;
const RUN_MAGIC: [u8; 8] = *b"HXWRUN02";
const REPLAY_MAGIC: [u8; 8] = *b"HXDEMO01";

const BUDGET_NONE: u8 = 0;
const BUDGET_STATE_ROWS: u8 = 1;
const BUDGET_EVENTS: u8 = 2;
const BUDGET_NESTED_ROWS: u8 = 3;
const BUDGET_GAS_FLOWS: u8 = 4;
const BUDGET_MULE_WAYPOINTS: u8 = 5;
const BUDGET_TOWN_LEDGER: u8 = 6;
const BUDGET_PACK_ITEMS: u8 = 7;
const BUDGET_HYMN_CARDS: u8 = 8;
const BUDGET_HYMN_MISCASTS: u8 = 9;
const BUDGET_LEDGER_DEBTS: u8 = 10;
const BUDGET_SHOP_ROWS: u8 = 11;
const BUDGET_ITEM_TAGS: u8 = 12;
const BUDGET_CHUNKS: u8 = 13;
const BUDGET_INITIAL_ROWS: u8 = 14;
const BUDGET_INITIAL_TAGS: u8 = 15;
const BUDGET_KNOWLEDGE_ROWS: u8 = 16;

#[derive(Default)]
struct DecodeBudget {
    strings: usize,
    chunks: usize,
    state_rows: usize,
    nested_rows: usize,
    gas_flows: usize,
    mule_waypoints: usize,
    town_ledger: usize,
    pack_items: usize,
    hymn_cards: usize,
    hymn_miscasts: usize,
    ledger_debts: usize,
    shop_rows: usize,
    item_tags: usize,
    initial_rows: usize,
    initial_tags: usize,
    initial_strings: usize,
    knowledge_rows: usize,
    knowledge_strings: usize,
}

impl DecodeBudget {
    fn add_counter(
        counter: &mut usize,
        len: usize,
        limit: usize,
        label: &str,
    ) -> Result<(), DecodeError> {
        *counter = counter
            .checked_add(len)
            .ok_or(DecodeError::Other("wire aggregate length overflow"))?;
        if *counter > limit {
            Err(DecodeError::OtherString(format!(
                "{label} aggregate {} exceeds limit {limit}",
                *counter
            )))
        } else {
            Ok(())
        }
    }

    fn add(&mut self, kind: u8, len: usize) -> Result<(), DecodeError> {
        let (counter, limit, label) = match kind {
            // Each envelope has exactly one event vector, so its field prefix
            // cap is also its aggregate cap; no second counter is necessary.
            BUDGET_NONE | BUDGET_EVENTS => return Ok(()),
            BUDGET_CHUNKS => (&mut self.chunks, MAX_TOTAL_CHUNKS, "total chunks"),
            BUDGET_STATE_ROWS => (
                &mut self.state_rows,
                MAX_TOTAL_STATE_ROWS,
                "total state rows",
            ),
            BUDGET_NESTED_ROWS => (
                &mut self.nested_rows,
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            ),
            BUDGET_GAS_FLOWS => (&mut self.gas_flows, MAX_TOTAL_GAS_FLOWS, "spill gas flows"),
            BUDGET_MULE_WAYPOINTS => (
                &mut self.mule_waypoints,
                MAX_TOTAL_MULE_WAYPOINTS,
                "mule waypoints",
            ),
            BUDGET_TOWN_LEDGER => (
                &mut self.town_ledger,
                MAX_TOTAL_TOWN_LEDGER_ROWS,
                "reputation town-ledger rows",
            ),
            BUDGET_PACK_ITEMS => (&mut self.pack_items, MAX_TOTAL_PACK_ITEMS, "pack items"),
            BUDGET_HYMN_CARDS => (&mut self.hymn_cards, MAX_TOTAL_HYMN_CARDS, "hymn cards"),
            BUDGET_HYMN_MISCASTS => (
                &mut self.hymn_miscasts,
                MAX_TOTAL_HYMN_MISCASTS,
                "hymn miscasts",
            ),
            BUDGET_LEDGER_DEBTS => (
                &mut self.ledger_debts,
                MAX_TOTAL_LEDGER_DEBTS,
                "ledger debts",
            ),
            BUDGET_SHOP_ROWS => (&mut self.shop_rows, MAX_TOTAL_SHOP_ROWS, "shop memory rows"),
            BUDGET_ITEM_TAGS => (&mut self.item_tags, MAX_TOTAL_ITEM_TAGS, "item tags"),
            BUDGET_INITIAL_ROWS => (
                &mut self.initial_rows,
                MAX_INITIAL_ROWS,
                "initial lineage rows",
            ),
            BUDGET_INITIAL_TAGS => (
                &mut self.initial_tags,
                MAX_INITIAL_TAGS,
                "initial lineage tags",
            ),
            BUDGET_KNOWLEDGE_ROWS => (
                &mut self.knowledge_rows,
                KNOWLEDGE_TOTAL_ROW_LIMIT,
                "working knowledge rows",
            ),
            _ => return Err(DecodeError::Other("unknown wire budget category")),
        };
        Self::add_counter(counter, len, limit, label)?;
        if matches!(
            kind,
            BUDGET_CHUNKS | BUDGET_INITIAL_ROWS | BUDGET_KNOWLEDGE_ROWS
        ) {
            Self::add_counter(
                &mut self.state_rows,
                len,
                MAX_TOTAL_STATE_ROWS,
                "total state rows",
            )?;
        }
        if matches!(kind, BUDGET_INITIAL_ROWS | BUDGET_INITIAL_TAGS) {
            Self::add_counter(
                &mut self.nested_rows,
                len,
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
        }
        if !matches!(
            kind,
            BUDGET_STATE_ROWS
                | BUDGET_NESTED_ROWS
                | BUDGET_CHUNKS
                | BUDGET_INITIAL_ROWS
                | BUDGET_INITIAL_TAGS
                | BUDGET_KNOWLEDGE_ROWS
        ) {
            Self::add_counter(
                &mut self.nested_rows,
                len,
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
        }
        Ok(())
    }

    fn add_strings(&mut self, len: usize) -> Result<(), DecodeError> {
        Self::add_counter(
            &mut self.strings,
            len,
            MAX_TOTAL_STRING_BYTES,
            "total strings",
        )
    }

    fn add_initial_string(&mut self, len: usize) -> Result<(), DecodeError> {
        Self::add_counter(
            &mut self.initial_strings,
            len,
            MAX_INITIAL_STRING_BYTES,
            "initial lineage strings",
        )
    }
    fn add_knowledge_string(&mut self, len: usize) -> Result<(), DecodeError> {
        Self::add_counter(
            &mut self.knowledge_strings,
            len,
            KNOWLEDGE_TOTAL_ID_BYTES_LIMIT,
            "working knowledge strings",
        )
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
struct BoundedString(String);

impl Deref for BoundedString {
    type Target = str;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}
impl DerefMut for BoundedString {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}
impl From<String> for BoundedString {
    fn from(value: String) -> Self {
        Self(value)
    }
}
impl From<&str> for BoundedString {
    fn from(value: &str) -> Self {
        Self(value.to_owned())
    }
}
impl From<BoundedString> for String {
    fn from(value: BoundedString) -> Self {
        value.0
    }
}
impl PartialEq<str> for BoundedString {
    fn eq(&self, other: &str) -> bool {
        self.0 == other
    }
}
impl PartialEq<&str> for BoundedString {
    fn eq(&self, other: &&str) -> bool {
        self.0 == *other
    }
}
impl BoundedString {
    fn into_inner(self) -> String {
        self.0
    }
}
impl Encode for BoundedString {
    fn encode<E: bincode::enc::Encoder>(
        &self,
        encoder: &mut E,
    ) -> Result<(), bincode::error::EncodeError> {
        (self.0.len() as u64).encode(encoder)?;
        bincode::enc::write::Writer::write(encoder.writer(), self.0.as_bytes())
    }
}
impl Decode<DecodeBudget> for BoundedString {
    fn decode<D: Decoder<Context = DecodeBudget>>(decoder: &mut D) -> Result<Self, DecodeError> {
        decode_string_with_cap(decoder, MAX_STRING_BYTES, false)
    }
}
bincode::impl_borrow_decode_with_context!(BoundedString, DecodeBudget);

fn decode_string_with_cap<D: Decoder<Context = DecodeBudget>>(
    decoder: &mut D,
    cap: usize,
    initial: bool,
) -> Result<BoundedString, DecodeError> {
    let raw_len = u64::decode(decoder)?;
    let len = usize::try_from(raw_len).map_err(|_| DecodeError::OutsideUsizeRange(raw_len))?;
    if len > cap {
        return Err(DecodeError::OtherString(format!(
            "wire string length {len} exceeds per-field limit {cap}"
        )));
    }
    decoder.context().add_strings(len)?;
    if initial {
        decoder.context().add_initial_string(len)?;
    }
    decoder.claim_container_read::<u8>(len)?;
    let mut bytes = vec![0; len];
    bincode::de::read::Reader::read(decoder.reader(), &mut bytes)?;
    String::from_utf8(bytes)
        .map(BoundedString)
        .map_err(|e| DecodeError::Utf8 {
            inner: e.utf8_error(),
        })
}

#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
struct InitialString(BoundedString);
impl Encode for InitialString {
    fn encode<E: bincode::enc::Encoder>(
        &self,
        encoder: &mut E,
    ) -> Result<(), bincode::error::EncodeError> {
        self.0.encode(encoder)
    }
}
impl Decode<DecodeBudget> for InitialString {
    fn decode<D: Decoder<Context = DecodeBudget>>(decoder: &mut D) -> Result<Self, DecodeError> {
        decode_string_with_cap(decoder, KNOWLEDGE_ID_MAX_BYTES, true).map(Self)
    }
}
bincode::impl_borrow_decode_with_context!(InitialString, DecodeBudget);
impl From<String> for InitialString {
    fn from(value: String) -> Self {
        Self(value.into())
    }
}
impl InitialString {
    fn into_inner(self) -> String {
        self.0.into_inner()
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
struct InitialText<const CAP: usize>(BoundedString);
impl<const CAP: usize> Encode for InitialText<CAP> {
    fn encode<E: bincode::enc::Encoder>(
        &self,
        encoder: &mut E,
    ) -> Result<(), bincode::error::EncodeError> {
        self.0.encode(encoder)
    }
}
impl<const CAP: usize> Decode<DecodeBudget> for InitialText<CAP> {
    fn decode<D: Decoder<Context = DecodeBudget>>(decoder: &mut D) -> Result<Self, DecodeError> {
        decode_string_with_cap(decoder, CAP, true).map(Self)
    }
}
impl<'de, const CAP: usize> bincode::BorrowDecode<'de, DecodeBudget> for InitialText<CAP> {
    fn borrow_decode<D: bincode::de::BorrowDecoder<'de, Context = DecodeBudget>>(
        decoder: &mut D,
    ) -> Result<Self, DecodeError> {
        Self::decode(decoder)
    }
}
impl<const CAP: usize> From<String> for InitialText<CAP> {
    fn from(value: String) -> Self {
        Self(value.into())
    }
}
impl<const CAP: usize> InitialText<CAP> {
    fn into_inner(self) -> String {
        self.0.into_inner()
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
struct KnowledgeString(BoundedString);
impl Encode for KnowledgeString {
    fn encode<E: bincode::enc::Encoder>(
        &self,
        encoder: &mut E,
    ) -> Result<(), bincode::error::EncodeError> {
        self.0.encode(encoder)
    }
}
impl Decode<DecodeBudget> for KnowledgeString {
    fn decode<D: Decoder<Context = DecodeBudget>>(decoder: &mut D) -> Result<Self, DecodeError> {
        let value = decode_string_with_cap(decoder, KNOWLEDGE_ID_MAX_BYTES, false)?;
        decoder.context().add_knowledge_string(value.len())?;
        Ok(Self(value))
    }
}
bincode::impl_borrow_decode_with_context!(KnowledgeString, DecodeBudget);
impl From<String> for KnowledgeString {
    fn from(value: String) -> Self {
        Self(value.into())
    }
}
impl KnowledgeString {
    fn into_inner(self) -> String {
        self.0.into_inner()
    }
}

#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct BoundedVec<T, const LIMIT: usize, const BUDGET: u8>(Vec<T>);

impl<T, const LIMIT: usize, const BUDGET: u8> Default for BoundedVec<T, LIMIT, BUDGET> {
    fn default() -> Self {
        Self(Vec::new())
    }
}

impl<T, const LIMIT: usize, const BUDGET: u8> Deref for BoundedVec<T, LIMIT, BUDGET> {
    type Target = [T];
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}
impl<T, const LIMIT: usize, const BUDGET: u8> DerefMut for BoundedVec<T, LIMIT, BUDGET> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}
impl<T, const LIMIT: usize, const BUDGET: u8> From<Vec<T>> for BoundedVec<T, LIMIT, BUDGET> {
    fn from(value: Vec<T>) -> Self {
        Self(value)
    }
}
impl<T, const LIMIT: usize, const BUDGET: u8> BoundedVec<T, LIMIT, BUDGET> {
    fn into_inner(self) -> Vec<T> {
        self.0
    }
}
impl<T, const LIMIT: usize, const BUDGET: u8> FromIterator<T> for BoundedVec<T, LIMIT, BUDGET> {
    fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
        Self(iter.into_iter().collect())
    }
}
impl<T, const LIMIT: usize, const BUDGET: u8> IntoIterator for BoundedVec<T, LIMIT, BUDGET> {
    type Item = T;
    type IntoIter = std::vec::IntoIter<T>;
    fn into_iter(self) -> Self::IntoIter {
        self.0.into_iter()
    }
}
impl<'a, T, const LIMIT: usize, const BUDGET: u8> IntoIterator
    for &'a BoundedVec<T, LIMIT, BUDGET>
{
    type Item = &'a T;
    type IntoIter = std::slice::Iter<'a, T>;
    fn into_iter(self) -> Self::IntoIter {
        self.0.iter()
    }
}
impl<T: Encode, const LIMIT: usize, const BUDGET: u8> Encode for BoundedVec<T, LIMIT, BUDGET> {
    fn encode<E: bincode::enc::Encoder>(
        &self,
        encoder: &mut E,
    ) -> Result<(), bincode::error::EncodeError> {
        (self.0.len() as u64).encode(encoder)?;
        for item in &self.0 {
            item.encode(encoder)?;
        }
        Ok(())
    }
}
impl<T: Decode<DecodeBudget>, const LIMIT: usize, const BUDGET: u8> Decode<DecodeBudget>
    for BoundedVec<T, LIMIT, BUDGET>
{
    fn decode<D: Decoder<Context = DecodeBudget>>(decoder: &mut D) -> Result<Self, DecodeError> {
        let raw_len = u64::decode(decoder)?;
        let len = usize::try_from(raw_len).map_err(|_| DecodeError::OutsideUsizeRange(raw_len))?;
        if len > LIMIT {
            return Err(DecodeError::OtherString(format!(
                "wire vector length {len} exceeds per-field limit {LIMIT}"
            )));
        }
        decoder.context().add(BUDGET, len)?;
        decoder.claim_container_read::<T>(len)?;
        let mut values = Vec::with_capacity(len);
        for _ in 0..len {
            decoder.unclaim_bytes_read(std::mem::size_of::<T>());
            values.push(T::decode(decoder)?);
        }
        Ok(Self(values))
    }
}
impl<'de, T, const LIMIT: usize, const BUDGET: u8> bincode::BorrowDecode<'de, DecodeBudget>
    for BoundedVec<T, LIMIT, BUDGET>
where
    T: Decode<DecodeBudget>,
{
    fn borrow_decode<D: bincode::de::BorrowDecoder<'de, Context = DecodeBudget>>(
        decoder: &mut D,
    ) -> Result<Self, DecodeError> {
        Self::decode(decoder)
    }
}

/// Immutable canonical inputs needed to reconstruct a run.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RunIdentity {
    pub seed: u64,
    pub mode: Mode,
    pub ancestry: Ancestry,
    pub calling: Calling,
    pub display_name: String,
    pub content_hash: String,
    pub lineage_id: LineageId,
    pub run_id: RunId,
    pub initial_lineage: InitialLineage,
}

/// A complete explicit core image with its independently verified hash.
#[derive(Clone, Debug)]
pub struct HashedState {
    pub tick: u64,
    pub hash: u64,
    pub image: StateImage,
}

impl HashedState {
    /// Capture and hash a validated live State without serializing core types.
    pub fn from_state(state: &State) -> Result<Self, PersistenceError> {
        let image = state
            .to_image()
            .map_err(|e| PersistenceError::CoreState(format!("{e:?}")))?;
        let hash = state
            .try_hash_state()
            .map_err(|e| PersistenceError::CoreState(format!("{e:?}")))?;
        Ok(Self {
            tick: image.tick,
            hash,
            image,
        })
    }

    /// Validate an owned image through core and attach its recomputed hash.
    pub fn from_image(image: StateImage) -> Result<Self, PersistenceError> {
        let tick = image.tick;
        let state = State::from_image(image.clone())
            .map_err(|e| PersistenceError::CoreImage(format!("{e:?}")))?;
        let hash = state
            .try_hash_state()
            .map_err(|e| PersistenceError::CoreState(format!("{e:?}")))?;
        Ok(Self { tick, hash, image })
    }

    /// Consume this checked image and restore the live core State.
    pub fn into_state(self) -> Result<State, PersistenceError> {
        validate_hashed_image(&self, "hashed_state")?;
        State::from_image(self.image).map_err(|e| PersistenceError::CoreImage(format!("{e:?}")))
    }
}

/// Nonrecursive Story checkpoint active at a retained event prefix.
#[derive(Clone, Debug)]
pub struct StoryCheckpoint {
    pub state: HashedState,
    pub after_event_count: u32,
}

/// Ordered lifecycle row. Sequence authority is the vector index, not a tick.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SessionEvent {
    pub pre_tick: u64,
    pub post_tick: u64,
    pub kind: SessionEventKind,
}

impl SessionEvent {
    #[must_use]
    pub fn action(pre_tick: u64, post_tick: u64, action: CommittedAction) -> Self {
        Self {
            pre_tick,
            post_tick,
            kind: SessionEventKind::Action(action),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SessionEventKind {
    Action(CommittedAction),
    StoryResurrect {
        checkpoint_after_event_count: u32,
        checkpoint_hash: u64,
        dead_state_hash: u64,
        restored_state_hash: u64,
    },
}

/// Cheap checked replay-work accounting. Session-owned trusted limits compare
/// their ceilings to these exact costs before world construction or stepping.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct TimelineCosts {
    pub full_steps: u64,
    pub unattended_steps: u64,
    pub total_elapsed_ticks: u64,
}

/// Decoded authoritative run. Current State is restored exactly; retained events
/// and the optional Story checkpoint are not replayed during ordinary resume.
#[derive(Debug)]
pub struct DecodedRun {
    pub identity: RunIdentity,
    pub current: State,
    pub story_checkpoint: Option<StoryCheckpoint>,
    pub events: Vec<SessionEvent>,
}

/// Production replay input. It intentionally contains no snapshot, checkpoint,
/// subtitle list, or duplicated rule-chain rows.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ProductionReplay {
    pub identity: RunIdentity,
    pub events: Vec<SessionEvent>,
    pub expected_final_tick: u64,
    pub expected_final_hash: u64,
}

/// Structured failure at the untrusted persistence boundary.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PersistenceError {
    InputTooLarge {
        actual: usize,
        limit: usize,
    },
    DecompressedTooLarge {
        limit: usize,
    },
    Compression(String),
    Decompression(String),
    Decode(String),
    TrailingBytes {
        consumed: usize,
        total: usize,
    },
    WrongMagic {
        format: &'static str,
    },
    OlderVersion {
        format: &'static str,
        found: u32,
        current: u32,
    },
    NewerVersion {
        format: &'static str,
        found: u32,
        current: u32,
    },
    CanonicalHashVersion {
        found: u64,
        current: u64,
    },
    ContentHashMismatch {
        stored: String,
        current: String,
    },
    MetadataMismatch {
        field: &'static str,
        envelope: u64,
        image: u64,
    },
    CheckpointHashMismatch {
        expected: u64,
        actual: u64,
    },
    Limit {
        context: String,
        actual: usize,
        limit: usize,
    },
    Invalid {
        context: String,
    },
    CoreImage(String),
    CoreState(String),
}

impl fmt::Display for PersistenceError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InputTooLarge { actual, limit } => write!(f, "compressed input is {actual} bytes; limit is {limit}"),
            Self::DecompressedTooLarge { limit } => write!(f, "decompressed input exceeds {limit} bytes"),
            Self::Compression(e) => write!(f, "compression failed: {e}"),
            Self::Decompression(e) => write!(f, "decompression failed: {e}"),
            Self::Decode(e) => write!(f, "bounded bincode decode failed: {e}"),
            Self::TrailingBytes { consumed, total } => write!(f, "trailing bytes after record: consumed {consumed} of {total}"),
            Self::WrongMagic { format } => write!(f, "input is not a {format} envelope"),
            Self::OlderVersion { format, found, current } => write!(f, "{format} version {found} is older than supported version {current}; no silent migration exists"),
            Self::NewerVersion { format, found, current } => write!(f, "{format} version {found} is newer than supported version {current}; update HEXWORK"),
            Self::CanonicalHashVersion { found, current } => write!(f, "core canonical-hash version {found} does not match current version {current}"),
            Self::ContentHashMismatch { stored, current } => write!(f, "content hash mismatch: stored {stored}, current {current}"),
            Self::MetadataMismatch { field, envelope, image } => write!(f, "run envelope {field} {envelope} does not match checkpoint image {image}"),
            Self::CheckpointHashMismatch { expected, actual } => write!(f, "checkpoint canonical hash mismatch: expected {expected}, recomputed {actual}"),
            Self::Limit { context, actual, limit } => write!(f, "{context} has {actual} entries/bytes; limit is {limit}"),
            Self::Invalid { context } => write!(f, "invalid persistence field: {context}"),
            Self::CoreImage(e) => write!(f, "core rejected checkpoint image: {e}"),
            Self::CoreState(e) => write!(f, "core state validation failed: {e}"),
        }
    }
}

impl std::error::Error for PersistenceError {}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct EnvelopeHeader {
    magic: [u8; 8],
    format_version: u32,
    canonical_hash_version: u64,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct RunWire {
    header: EnvelopeHeader,
    identity: RunIdentityWire,
    current: HashedStateWire,
    story_checkpoint: Option<StoryCheckpointWire>,
    events: BoundedVec<SessionEventWire, MAX_SESSION_EVENTS, BUDGET_EVENTS>,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ReplayWire {
    header: EnvelopeHeader,
    identity: RunIdentityWire,
    events: BoundedVec<SessionEventWire, MAX_SESSION_EVENTS, BUDGET_EVENTS>,
    expected_final_tick: u64,
    expected_final_hash: u64,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct RunIdentityWire {
    seed: u64,
    mode: u8,
    ancestry: u8,
    calling: u8,
    display_name: BoundedString,
    content_hash: BoundedString,
    lineage_id: [u8; 16],
    run_sequence: u64,
    initial_lineage: InitialLineageWire,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct InitialLineageWire {
    mapped_regions: BoundedVec<InitialString, MAX_MAPPED_REGIONS, BUDGET_INITIAL_ROWS>,
    recipes: BoundedVec<InitialString, MAX_RECIPES, BUDGET_INITIAL_ROWS>,
    translated_hymns: BoundedVec<InitialString, MAX_TRANSLATIONS, BUDGET_INITIAL_ROWS>,
    identified_item_templates:
        BoundedVec<InitialString, MAX_IDENTIFIED_TEMPLATES, BUDGET_INITIAL_ROWS>,
    ghosts: BoundedVec<GhostWire, MAX_GHOSTS, BUDGET_INITIAL_ROWS>,
    auction_lots: BoundedVec<AuctionLotWire, MAX_AUCTION_LOTS, BUDGET_INITIAL_ROWS>,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct GhostWire {
    origin_run_sequence: u64,
    item_template_id: InitialString,
    warning: InitialText<MAX_GHOST_WARNING_BYTES>,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct AuctionLotWire {
    lot_run_sequence: u64,
    lot_item_id: u64,
    origin_run_sequence: u64,
    origin_item_id: u64,
    template_id: InitialString,
    material: u8,
    identified: bool,
    cursed: bool,
    blessed: bool,
    condition: u8,
    silvered: bool,
    illegal: bool,
    tags: BoundedVec<InitialText<MAX_LOT_TAG_BYTES>, MAX_LOT_TAGS, BUDGET_INITIAL_TAGS>,
    fuel: u8,
    wet_ticks: u64,
    base_price_cu: u32,
    buyback_price_cu: u32,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct HashedStateWire {
    tick: u64,
    hash: u64,
    image: StateWire,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct StoryCheckpointWire {
    state: HashedStateWire,
    after_event_count: u32,
}

#[derive(Clone, Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct SessionEventWire {
    pre_tick: u64,
    post_tick: u64,
    kind_tag: u8,
    action: ActionWire,
    checkpoint_after_event_count: u32,
    checkpoint_hash: u64,
    dead_state_hash: u64,
    restored_state_hash: u64,
}

#[derive(Clone, Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ActionWire {
    tag: u8,
    direction: u8,
    service: u8,
    text: BoundedString,
    id: u64,
}

/// One pending-action row: operation tag, slot counters, and a typed union
/// payload. The `tag` selects which payload fields are canonical; every other
/// field must be its zero/default value (checked decoding).
#[derive(Clone, Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct PendingRowWire {
    tag: u8,
    start_tick: u64,
    remaining_ticks: u64,
    direction: u8,
    service: u8,
    text: BoundedString,
    id: u64,
    inn_total: u64,
    inn_remaining: u64,
}

/// One terrain-observation row (docs/fluid §5): location provenance, the voxel
/// boundary face, and the last-seen appearance plus observation tick and branch
/// epoch. There is deliberately no actor/item/service identity field.
#[derive(Clone, Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ObservationRowWire {
    location_kind: u8,
    town: BoundedString,
    depth: u8,
    cell: CellWire,
    face: u8,
    material: u8,
    occupancy: u8,
    door_appearance: u8,
    observed_tick: u64,
    branch_epoch: u64,
}

#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct StateWire {
    tick: u64,
    seed: u64,
    rng_master: [u64; 4],
    ecs_entity_count: u32,
    chunks: BoundedVec<ChunkWire, MAX_CHUNKS, BUDGET_CHUNKS>,
    actors: BoundedVec<ActorWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    identities: BoundedVec<(u64, IdentityWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    wallets: BoundedVec<(u64, WalletWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    contracts: BoundedVec<(BoundedString, ContractWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    location: LocationWire,
    connectors: BoundedVec<ConnectorWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    pressure_nodes: BoundedVec<PressureNodeWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    pressure_edges: BoundedVec<PressureEdgeWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    causes: BoundedVec<CauseWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    far: FarWire,
    laws: BoundedVec<LawWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    fall_odo: BoundedVec<(u64, u64), MAX_ROWS, BUDGET_STATE_ROWS>,
    lights: BoundedVec<LightWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    sounds: BoundedVec<SoundWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    corpses: BoundedVec<(CellWire, CorpseWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    signal_sources: BoundedVec<CellWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    spill: SpillWire,
    characters: BoundedVec<(u64, CharacterWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    reputations: BoundedVec<(u64, ReputationWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    items: BoundedVec<(u64, ItemWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    ground_items: BoundedVec<(u64, CellWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    packs: BoundedVec<(u64, PackWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    knowledge: KnowledgeWire,
    idf: BoundedVec<(BoundedString, u8), MAX_ROWS, BUDGET_STATE_ROWS>,
    hymn_cases: BoundedVec<(u64, HymnCaseWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    cast_queue: BoundedVec<CastWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    bargains: BoundedVec<(u64, LedgerWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    deep_weather: i8,
    npcs: BoundedVec<(u64, NpcWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    shops: BoundedVec<(CellWire, ShopWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    gossip: BoundedVec<GossipWire, MAX_ROWS, BUDGET_STATE_ROWS>,
    chalk: BoundedVec<(CellWire, BoundedString), MAX_ROWS, BUDGET_STATE_ROWS>,
    mule: Option<BoundedVec<CellWire, MAX_TOTAL_MULE_WAYPOINTS, BUDGET_MULE_WAYPOINTS>>,
    vault_choice: u8,
    chapel_debts: BoundedVec<(u64, ChapelDebtWire), MAX_ROWS, BUDGET_STATE_ROWS>,
    next_id: u64,
    step_mode: bool,
    /// The single pending-action slot as a bounded, tagged row vector (0 or 1
    /// rows). A second row is a duplicate-slot violation, rejected on decode.
    pending: BoundedVec<PendingRowWire, MAX_PENDING_ROWS, BUDGET_STATE_ROWS>,
    /// Canonical player terrain-observation memory as a bounded, canonical-order
    /// row vector (docs/fluid §5). A duplicate or reversed row is rejected on
    /// decode.
    observations: BoundedVec<ObservationRowWire, MAX_OBSERVATION_ROWS, BUDGET_STATE_ROWS>,
    /// Durable Story branch epoch (docs/fluid §7). Appended last so the v5 wire
    /// stays append-only; it survives a save/reload and replay reproduces the
    /// same fold.
    branch_epoch: u64,
}

#[derive(Clone, Encode, Decode, PartialEq, Eq, PartialOrd, Ord)]
#[bincode(decode_context = "DecodeBudget")]
struct CellWire {
    x: i32,
    y: i32,
    z: i32,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ChunkWire {
    coord: CellWire,
    voxels: BoundedVec<VoxelWire, CHUNK_VOL, BUDGET_NESTED_ROWS>,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct VoxelWire {
    material: u8,
    occ: u8,
    fluid_kind: u8,
    fluid_amt: u8,
    gas_kind: u8,
    gas_amt: u8,
    light: u8,
    scent: u8,
    flags: u8,
    pad: u8,
    pressure: i16,
    flow: i16,
    extra: u16,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ActorWire {
    id: u64,
    position_mm: [i64; 3],
    body_mm: [i64; 3],
    heading: Option<u16>,
    movement_remainder: [i64; 2],
    stance: u8,
    hp: i32,
    alive: bool,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct IdentityWire {
    template: BoundedString,
    name: BoundedString,
    role: u8,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct WalletWire {
    copper: u32,
    scrip: u32,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ContractWire {
    id: BoundedString,
    taker: u64,
    accepted_tick: u64,
    floor: u8,
    pay_cu: u32,
    target: u32,
    progress: u32,
    status: u8,
    reward_town: BoundedString,
    standing_delta: i16,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct LocationWire {
    tag: u8,
    town: BoundedString,
    depth: u8,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ConnectorWire {
    id: u64,
    kind: u8,
    from: [u32; 3],
    to: [u32; 3],
    length: u32,
    anchor_quality: u8,
    load_kg: u32,
    blocked: bool,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct PressureNodeWire {
    id: u64,
    pressure: u32,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct PressureEdgeWire {
    from: u64,
    to: u64,
    openness: u32,
    hidden: bool,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct CauseWire {
    tick: u64,
    rule: BoundedString,
    detail: BoundedString,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct FarWire {
    day: u64,
    day_tick: u64,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct LawWire {
    rule: BoundedString,
    tick: u64,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct LightWire {
    x: i32,
    y: i32,
    z: i32,
    radius: u8,
    warm: bool,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct SoundWire {
    x: i32,
    y: i32,
    z: i32,
    loudness: u8,
    timbre: u8,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct CorpseWire {
    age_ticks: u64,
    bursts_at: u64,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct SpillWire {
    desired_moves: BoundedVec<(u64, Option<u16>, u32, Option<[i64;3]>), MAX_ROWS, BUDGET_STATE_ROWS>,
    crossed_cells: BoundedVec<(u64, BoundedVec<CellWire, MAX_ROWS, BUDGET_NESTED_ROWS>), MAX_ROWS, BUDGET_STATE_ROWS>,
    gas_flows: BoundedVec<(u64, u64, u32), MAX_ROWS, BUDGET_GAS_FLOWS>,
    light_pending: bool,
    signal_pending: bool,
    scent_pending: bool,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct CharacterWire {
    ancestry: u8,
    calling: u8,
    attrs: [u8; 5],
    skills: [u8; 14],
    uses: [u32; 14],
    pips: [u8; 14],
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct StandingWire {
    faction: u8,
    rank: i8,
    debt_cu: i32,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct GodCreditWire {
    god: u8,
    credit: i16,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ReputationWire {
    factions: [StandingWire; 5],
    town_ledger: BoundedVec<(BoundedString, i16), MAX_TOWN_LEDGER_ROWS, BUDGET_TOWN_LEDGER>,
    god_credit: [GodCreditWire; 5],
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ItemWire {
    id: u64,
    template: BoundedString,
    material: u8,
    identified: bool,
    cursed: bool,
    blessed: bool,
    condition: u8,
    silvered: bool,
    illegal: bool,
    tags: BoundedVec<BoundedString, MAX_ITEM_TAGS, BUDGET_ITEM_TAGS>,
    fuel: u8,
    wet_ticks: u64,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct PackWire {
    items: BoundedVec<(u64, ItemWire), MAX_PACK_ITEMS, BUDGET_PACK_ITEMS>,
    equipped_weapon: Option<u64>,
    equipped_armor: Option<u64>,
    max_weight_kg: u32,
    max_bulk: u8,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct HymnCardWire {
    template: BoundedString,
    conduct_base: u8,
    slots: u8,
    fragile: bool,
    miscasts: BoundedVec<BoundedString, MAX_HYMN_MISCASTS, BUDGET_HYMN_MISCASTS>,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct HymnCaseWire {
    cards: BoundedVec<HymnCardWire, MAX_HYMN_CARDS, BUDGET_HYMN_CARDS>,
    max_slots: u8,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct CastWire {
    caster: u64,
    template: BoundedString,
    target: CellWire,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct DebtWire {
    god: u8,
    boon: BoundedString,
    owed: BoundedString,
    ticks_left: Option<u64>,
    altar_receipt: bool,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct LedgerWire {
    debts: BoundedVec<DebtWire, MAX_LEDGER_DEBTS, BUDGET_LEDGER_DEBTS>,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct NpcWire {
    name: BoundedString,
    job: u8,
    home: CellWire,
    work: CellWire,
    debt_cu: i32,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ShopWire {
    broke: BoundedVec<(u64, u8), MAX_SHOP_ROWS, BUDGET_SHOP_ROWS>,
    stole: BoundedVec<(u64, bool), MAX_SHOP_ROWS, BUDGET_SHOP_ROWS>,
    killed: BoundedVec<(u64, bool), MAX_SHOP_ROWS, BUDGET_SHOP_ROWS>,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct GossipWire {
    from: BoundedString,
    to: BoundedString,
    sent_day: u64,
    ledger_delta: i16,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
struct ChapelDebtWire {
    burials_owed: u8,
    chapel: BoundedString,
}
#[derive(Encode, Decode)]
#[bincode(decode_context = "DecodeBudget")]
pub struct RunKnowledgeWire {
    mapped_regions: BoundedVec<KnowledgeString, MAPPED_REGION_LIMIT, BUDGET_KNOWLEDGE_ROWS>,
    recipes: BoundedVec<KnowledgeString, RECIPE_LIMIT, BUDGET_KNOWLEDGE_ROWS>,
    translated_hymns: BoundedVec<KnowledgeString, TRANSLATION_LIMIT, BUDGET_KNOWLEDGE_ROWS>,
}
type KnowledgeWire = RunKnowledgeWire;

/// Encode an exact current State, immutable identity, optional nonrecursive
/// Story checkpoint, and the full retained lifecycle history.
pub fn encode_run(
    identity: &RunIdentity,
    current: &State,
    story_checkpoint: Option<&StoryCheckpoint>,
    events: &[SessionEvent],
) -> Result<Vec<u8>, PersistenceError> {
    let current_hashed = HashedState::from_state(current)?;
    validate_hashed_image(&current_hashed, "run.current")?;
    if let Some(checkpoint) = story_checkpoint {
        validate_hashed_image(&checkpoint.state, "run.story_checkpoint.state")?;
    }
    validate_run_structural_limits(
        identity,
        &current_hashed.image,
        story_checkpoint,
        events,
        None,
    )?;
    let file = RunWire {
        header: header(RUN_MAGIC, RUN_FORMAT_VERSION),
        identity: identity_to_wire(identity),
        current: hashed_state_to_wire(current_hashed)?,
        story_checkpoint: story_checkpoint.map(story_checkpoint_to_wire).transpose()?,
        events: events.iter().map(event_to_wire).collect::<Result<_, _>>()?,
    };
    encode_bounded(file)
}

/// Decode, integrity-check, reconstruct, validate, and hash a v4 run save.
pub fn decode_run(
    bytes: &[u8],
    current_content_hash: &str,
) -> Result<DecodedRun, PersistenceError> {
    valid_string(current_content_hash, "current_content_hash", true)?;
    let raw = decompress_bounded(bytes)?;
    gate_raw_header(&raw, RUN_MAGIC, "run-save", RUN_FORMAT_VERSION)?;
    let file: RunWire = decode_raw_bounded(&raw)?;
    let identity = identity_from_wire(file.identity, "run.identity")?;
    validate_content_hash(&identity.content_hash, current_content_hash)?;
    let current_hashed = hashed_state_from_wire(file.current, "run.current")?;
    let story_checkpoint = file
        .story_checkpoint
        .map(story_checkpoint_from_wire)
        .transpose()?;
    let events = wires_to_events(file.events.into_inner(), "run.events")?;
    validate_run_structural_limits(
        &identity,
        &current_hashed.image,
        story_checkpoint.as_ref(),
        &events,
        None,
    )?;
    let current = State::from_image(current_hashed.image)
        .map_err(|e| PersistenceError::CoreImage(format!("run.current: {e:?}")))?;
    Ok(DecodedRun {
        identity,
        current,
        story_checkpoint,
        events,
    })
}

/// Encode production replay input; no checkpoint, snapshot, or presentation
/// evidence is stored.
pub fn encode_replay(replay: &ProductionReplay) -> Result<Vec<u8>, PersistenceError> {
    validate_identity(&replay.identity, "replay.identity")?;
    validate_events(
        &replay.events,
        replay.identity.mode,
        replay.expected_final_tick,
        "replay.events",
    )?;
    let mut budget = EncodeBudget::default();
    budget.check_identity(&replay.identity, "replay.identity")?;
    budget.check_events(&replay.events, "replay.events")?;
    let wire = ReplayWire {
        header: header(REPLAY_MAGIC, REPLAY_FORMAT_VERSION),
        identity: identity_to_wire(&replay.identity),
        events: replay
            .events
            .iter()
            .map(event_to_wire)
            .collect::<Result<_, _>>()?,
        expected_final_tick: replay.expected_final_tick,
        expected_final_hash: replay.expected_final_hash,
    };
    encode_bounded(wire)
}

/// Decode and validate production replay input against the current content pack.
pub fn decode_replay(
    bytes: &[u8],
    current_content_hash: &str,
) -> Result<ProductionReplay, PersistenceError> {
    valid_string(current_content_hash, "current_content_hash", true)?;
    let raw = decompress_bounded(bytes)?;
    gate_raw_header(&raw, REPLAY_MAGIC, "replay", REPLAY_FORMAT_VERSION)?;
    let wire: ReplayWire = decode_raw_bounded(&raw)?;
    let identity = identity_from_wire(wire.identity, "replay.identity")?;
    validate_content_hash(&identity.content_hash, current_content_hash)?;
    let events = wires_to_events(wire.events.into_inner(), "replay.events")?;
    validate_events(
        &events,
        identity.mode,
        wire.expected_final_tick,
        "replay.events",
    )?;
    Ok(ProductionReplay {
        identity,
        events,
        expected_final_tick: wire.expected_final_tick,
        expected_final_hash: wire.expected_final_hash,
    })
}

fn encode_bounded<T: Encode>(value: T) -> Result<Vec<u8>, PersistenceError> {
    encode_bounded_with_limits(value, MAX_DECOMPRESSED_BYTES, MAX_COMPRESSED_BYTES)
}

fn encode_bounded_with_limits<T: Encode>(
    value: T,
    max_raw_bytes: usize,
    max_compressed_bytes: usize,
) -> Result<Vec<u8>, PersistenceError> {
    let raw = bincode::encode_to_vec(
        value,
        bincode::config::standard().with_limit::<MAX_DECOMPRESSED_BYTES>(),
    )
    .map_err(|e| PersistenceError::Decode(e.to_string()))?;
    if raw.len() > max_raw_bytes {
        return Err(PersistenceError::Limit {
            context: "encoded envelope".into(),
            actual: raw.len(),
            limit: max_raw_bytes,
        });
    }
    let compressed = zstd::stream::encode_all(raw.as_slice(), ZSTD_LEVEL)
        .map_err(|e| PersistenceError::Compression(e.to_string()))?;
    if compressed.len() > max_compressed_bytes {
        return Err(PersistenceError::InputTooLarge {
            actual: compressed.len(),
            limit: max_compressed_bytes,
        });
    }
    Ok(compressed)
}

fn decompress_bounded(bytes: &[u8]) -> Result<Vec<u8>, PersistenceError> {
    if bytes.len() > MAX_COMPRESSED_BYTES {
        return Err(PersistenceError::InputTooLarge {
            actual: bytes.len(),
            limit: MAX_COMPRESSED_BYTES,
        });
    }
    let decoder = zstd::stream::read::Decoder::new(bytes)
        .map_err(|e| PersistenceError::Decompression(e.to_string()))?;
    let mut raw = Vec::new();
    decoder
        .take((MAX_DECOMPRESSED_BYTES as u64) + 1)
        .read_to_end(&mut raw)
        .map_err(|e| PersistenceError::Decompression(e.to_string()))?;
    if raw.len() > MAX_DECOMPRESSED_BYTES {
        return Err(PersistenceError::DecompressedTooLarge {
            limit: MAX_DECOMPRESSED_BYTES,
        });
    }
    Ok(raw)
}

fn decode_raw_bounded<T: Decode<DecodeBudget>>(raw: &[u8]) -> Result<T, PersistenceError> {
    let (value, used) = bincode::decode_from_slice_with_context(
        raw,
        bincode::config::standard().with_limit::<MAX_DECOMPRESSED_BYTES>(),
        DecodeBudget::default(),
    )
    .map_err(|e| PersistenceError::Decode(e.to_string()))?;
    if used != raw.len() {
        return Err(PersistenceError::TrailingBytes {
            consumed: used,
            total: raw.len(),
        });
    }
    Ok(value)
}

fn gate_raw_header(
    raw: &[u8],
    magic: [u8; 8],
    format: &'static str,
    current: u32,
) -> Result<(), PersistenceError> {
    let (header, _) = bincode::decode_from_slice_with_context::<DecodeBudget, EnvelopeHeader, _>(
        raw,
        bincode::config::standard().with_limit::<MAX_DECOMPRESSED_BYTES>(),
        DecodeBudget::default(),
    )
    .map_err(|e| PersistenceError::Decode(e.to_string()))?;
    if header.magic != magic {
        return Err(PersistenceError::WrongMagic { format });
    }
    gate_version(format, header.format_version, current)?;
    gate_hash_version(header.canonical_hash_version)
}

fn gate_version(format: &'static str, found: u32, current: u32) -> Result<(), PersistenceError> {
    if found < current {
        Err(PersistenceError::OlderVersion {
            format,
            found,
            current,
        })
    } else if found > current {
        Err(PersistenceError::NewerVersion {
            format,
            found,
            current,
        })
    } else {
        Ok(())
    }
}
fn gate_hash_version(found: u64) -> Result<(), PersistenceError> {
    if found == CANONICAL_HASH_VERSION {
        Ok(())
    } else {
        Err(PersistenceError::CanonicalHashVersion {
            found,
            current: CANONICAL_HASH_VERSION,
        })
    }
}
fn metadata(envelope: u64, image: u64, field: &'static str) -> Result<(), PersistenceError> {
    if envelope == image {
        Ok(())
    } else {
        Err(PersistenceError::MetadataMismatch {
            field,
            envelope,
            image,
        })
    }
}
fn limit(actual: usize, limit: usize, context: &str) -> Result<(), PersistenceError> {
    if actual <= limit {
        Ok(())
    } else {
        Err(PersistenceError::Limit {
            context: context.into(),
            actual,
            limit,
        })
    }
}
fn valid_string(value: &str, context: &str, nonblank: bool) -> Result<(), PersistenceError> {
    limit(value.len(), MAX_STRING_BYTES, context)?;
    if nonblank && value.trim().is_empty() {
        Err(PersistenceError::Invalid {
            context: format!("{context} is blank"),
        })
    } else {
        Ok(())
    }
}

fn header(magic: [u8; 8], format_version: u32) -> EnvelopeHeader {
    EnvelopeHeader {
        magic,
        format_version,
        canonical_hash_version: CANONICAL_HASH_VERSION,
    }
}

fn validate_content_hash(stored: &str, current: &str) -> Result<(), PersistenceError> {
    if stored == current {
        Ok(())
    } else {
        Err(PersistenceError::ContentHashMismatch {
            stored: stored.to_owned(),
            current: current.to_owned(),
        })
    }
}

fn validate_identity(identity: &RunIdentity, context: &str) -> Result<(), PersistenceError> {
    valid_string(
        &identity.display_name,
        &format!("{context}.display_name"),
        true,
    )?;
    valid_string(
        &identity.content_hash,
        &format!("{context}.content_hash"),
        true,
    )?;
    validate_initial_lineage(&identity.initial_lineage, true).map_err(|e| {
        PersistenceError::Invalid {
            context: format!("{context}.initial_lineage: {e}"),
        }
    })
}

fn identity_to_wire(identity: &RunIdentity) -> RunIdentityWire {
    RunIdentityWire {
        seed: identity.seed,
        mode: mode_tag(identity.mode),
        ancestry: identity.ancestry.canonical_tag(),
        calling: identity.calling.canonical_tag(),
        display_name: identity.display_name.clone().into(),
        content_hash: identity.content_hash.clone().into(),
        lineage_id: *identity.lineage_id.as_bytes(),
        run_sequence: identity.run_id.sequence(),
        initial_lineage: initial_lineage_to_wire(&identity.initial_lineage),
    }
}

fn identity_from_wire(
    wire: RunIdentityWire,
    context: &str,
) -> Result<RunIdentity, PersistenceError> {
    let identity = RunIdentity {
        seed: wire.seed,
        mode: mode_from_tag(wire.mode, &format!("{context}.mode"))?,
        ancestry: ancestry(wire.ancestry, &format!("{context}.ancestry"))?,
        calling: calling(wire.calling, &format!("{context}.calling"))?,
        display_name: wire.display_name.into_inner(),
        content_hash: wire.content_hash.into_inner(),
        lineage_id: LineageId::from_bytes(wire.lineage_id),
        run_id: RunId::new(wire.run_sequence).map_err(|e| PersistenceError::Invalid {
            context: format!("{context}.run_id: {e}"),
        })?,
        initial_lineage: initial_lineage_from_wire(
            wire.initial_lineage,
            &format!("{context}.initial_lineage"),
        )?,
    };
    validate_identity(&identity, context)?;
    Ok(identity)
}

fn initial_lineage_to_wire(value: &InitialLineage) -> InitialLineageWire {
    InitialLineageWire {
        mapped_regions: value
            .mapped_regions
            .iter()
            .cloned()
            .map(Into::into)
            .collect(),
        recipes: value.recipes.iter().cloned().map(Into::into).collect(),
        translated_hymns: value
            .translated_hymns
            .iter()
            .cloned()
            .map(Into::into)
            .collect(),
        identified_item_templates: value
            .identified_item_templates
            .iter()
            .cloned()
            .map(Into::into)
            .collect(),
        ghosts: value
            .ghosts
            .iter()
            .map(|g| GhostWire {
                origin_run_sequence: g.origin_run_id.sequence(),
                item_template_id: g.item_template_id.clone().into(),
                warning: g.warning.clone().into(),
            })
            .collect(),
        auction_lots: value
            .auction_lots
            .iter()
            .map(|l| AuctionLotWire {
                lot_run_sequence: l.lot_id.origin_run_id().sequence(),
                lot_item_id: l.lot_id.origin_item_id(),
                origin_run_sequence: l.origin_run_id.sequence(),
                origin_item_id: l.origin_item_id,
                template_id: l.template_id.clone().into(),
                material: l.material,
                identified: l.identified,
                cursed: l.cursed,
                blessed: l.blessed,
                condition: l.condition.canonical_tag(),
                silvered: l.silvered,
                illegal: l.illegal,
                tags: l.tags.iter().cloned().map(Into::into).collect(),
                fuel: l.fuel,
                wet_ticks: l.wet_ticks,
                base_price_cu: l.base_price_cu,
                buyback_price_cu: l.buyback_price_cu,
            })
            .collect(),
    }
}
fn initial_lineage_from_wire(
    w: InitialLineageWire,
    context: &str,
) -> Result<InitialLineage, PersistenceError> {
    let ghosts = w
        .ghosts
        .into_iter()
        .map(|g| {
            Ok(GhostMemory {
                origin_run_id: RunId::new(g.origin_run_sequence).map_err(|e| {
                    PersistenceError::Invalid {
                        context: format!("{context}.ghost.run_id: {e}"),
                    }
                })?,
                item_template_id: g.item_template_id.into_inner(),
                warning: g.warning.into_inner(),
            })
        })
        .collect::<Result<_, PersistenceError>>()?;
    let lots = w
        .auction_lots
        .into_iter()
        .map(|l| {
            let origin_run_id =
                RunId::new(l.origin_run_sequence).map_err(|e| PersistenceError::Invalid {
                    context: format!("{context}.lot.run_id: {e}"),
                })?;
            let lot_run_id =
                RunId::new(l.lot_run_sequence).map_err(|e| PersistenceError::Invalid {
                    context: format!("{context}.lot.lot_id.run_id: {e}"),
                })?;
            let lot_id =
                LotId::new(lot_run_id, l.lot_item_id).map_err(|e| PersistenceError::Invalid {
                    context: format!("{context}.lot.lot_id: {e}"),
                })?;
            if lot_id
                != LotId::new(origin_run_id, l.origin_item_id).map_err(|e| {
                    PersistenceError::Invalid {
                        context: format!("{context}.lot.origin: {e}"),
                    }
                })?
            {
                return Err(PersistenceError::Invalid {
                    context: format!("{context}.lot.lot_id does not match origin run/item"),
                });
            }
            Ok(AuctionLot {
                lot_id,
                origin_run_id,
                origin_item_id: l.origin_item_id,
                template_id: l.template_id.into_inner(),
                material: l.material,
                identified: l.identified,
                cursed: l.cursed,
                blessed: l.blessed,
                condition: condition(l.condition, &format!("{context}.lot.condition"))?,
                silvered: l.silvered,
                illegal: l.illegal,
                tags: l.tags.into_iter().map(InitialText::into_inner).collect(),
                fuel: l.fuel,
                wet_ticks: l.wet_ticks,
                base_price_cu: l.base_price_cu,
                buyback_price_cu: l.buyback_price_cu,
            })
        })
        .collect::<Result<_, PersistenceError>>()?;
    let value = InitialLineage {
        mapped_regions: w
            .mapped_regions
            .into_iter()
            .map(InitialString::into_inner)
            .collect(),
        recipes: w
            .recipes
            .into_iter()
            .map(InitialString::into_inner)
            .collect(),
        translated_hymns: w
            .translated_hymns
            .into_iter()
            .map(InitialString::into_inner)
            .collect(),
        identified_item_templates: w
            .identified_item_templates
            .into_iter()
            .map(InitialString::into_inner)
            .collect(),
        ghosts,
        auction_lots: lots,
    };
    validate_initial_lineage(&value, true).map_err(|e| PersistenceError::Invalid {
        context: format!("{context}: {e}"),
    })?;
    Ok(value)
}
fn knowledge_to_wire(value: RunKnowledge) -> KnowledgeWire {
    KnowledgeWire {
        mapped_regions: value.mapped_regions.into_iter().map(Into::into).collect(),
        recipes: value.recipes.into_iter().map(Into::into).collect(),
        translated_hymns: value.translated_hymns.into_iter().map(Into::into).collect(),
    }
}
fn require_strict_knowledge_order<const LIMIT: usize>(
    rows: &BoundedVec<KnowledgeString, LIMIT, BUDGET_KNOWLEDGE_ROWS>,
    context: &str,
) -> Result<(), PersistenceError> {
    if rows.windows(2).any(|pair| pair[0] >= pair[1]) {
        Err(PersistenceError::Invalid {
            context: format!("{context} must be strictly increasing and unique"),
        })
    } else {
        Ok(())
    }
}

fn knowledge_from_wire(value: KnowledgeWire) -> Result<RunKnowledge, PersistenceError> {
    require_strict_knowledge_order(&value.mapped_regions, "state.knowledge.mapped_regions")?;
    require_strict_knowledge_order(&value.recipes, "state.knowledge.recipes")?;
    require_strict_knowledge_order(&value.translated_hymns, "state.knowledge.translated_hymns")?;
    Ok(RunKnowledge {
        mapped_regions: value
            .mapped_regions
            .into_iter()
            .map(KnowledgeString::into_inner)
            .collect(),
        recipes: value
            .recipes
            .into_iter()
            .map(KnowledgeString::into_inner)
            .collect(),
        translated_hymns: value
            .translated_hymns
            .into_iter()
            .map(KnowledgeString::into_inner)
            .collect(),
    })
}

fn hashed_state_to_wire(state: HashedState) -> Result<HashedStateWire, PersistenceError> {
    Ok(HashedStateWire {
        tick: state.tick,
        hash: state.hash,
        image: image_to_wire(state.image)?,
    })
}

fn hashed_state_from_wire(
    wire: HashedStateWire,
    context: &str,
) -> Result<HashedState, PersistenceError> {
    let image = wire_to_image(wire.image)?;
    metadata(wire.tick, image.tick, "tick")?;
    let restored = State::from_image(image.clone())
        .map_err(|e| PersistenceError::CoreImage(format!("{context}: {e:?}")))?;
    let actual = restored
        .try_hash_state()
        .map_err(|e| PersistenceError::CoreState(format!("{context}: {e:?}")))?;
    if actual != wire.hash {
        return Err(PersistenceError::CheckpointHashMismatch {
            expected: wire.hash,
            actual,
        });
    }
    Ok(HashedState {
        tick: wire.tick,
        hash: wire.hash,
        image,
    })
}

fn story_checkpoint_to_wire(
    checkpoint: &StoryCheckpoint,
) -> Result<StoryCheckpointWire, PersistenceError> {
    Ok(StoryCheckpointWire {
        state: hashed_state_to_wire(checkpoint.state.clone())?,
        after_event_count: checkpoint.after_event_count,
    })
}

fn story_checkpoint_from_wire(
    wire: StoryCheckpointWire,
) -> Result<StoryCheckpoint, PersistenceError> {
    Ok(StoryCheckpoint {
        state: hashed_state_from_wire(wire.state, "run.story_checkpoint.state")?,
        after_event_count: wire.after_event_count,
    })
}

fn validate_hashed_image(state: &HashedState, context: &str) -> Result<(), PersistenceError> {
    metadata(state.tick, state.image.tick, "tick")?;
    let restored = State::from_image(state.image.clone())
        .map_err(|e| PersistenceError::CoreImage(format!("{context}: {e:?}")))?;
    let actual = restored
        .try_hash_state()
        .map_err(|e| PersistenceError::CoreState(format!("{context}: {e:?}")))?;
    if actual != state.hash {
        return Err(PersistenceError::CheckpointHashMismatch {
            expected: state.hash,
            actual,
        });
    }
    Ok(())
}

/// Exact journal span of one committed operation, sourced from the core
/// [`hexwork_core::action_timing::action_timing`] table — the single source of
/// truth for durations (docs/fluid §4.2). Returns `(world_ticks, unattended_ticks)`.
fn action_semantic_cost(action: &CommittedAction) -> Result<(u64, u64), PersistenceError> {
    let timing = hexwork_core::action_timing::action_timing(action);
    Ok(match timing {
        hexwork_core::action_timing::ActionTiming::Observation => (0, 0),
        hexwork_core::action_timing::ActionTiming::World(ticks) => (ticks.get(), 0),
        hexwork_core::action_timing::ActionTiming::Unattended(ticks) => (0, ticks.get()),
    })
}

fn checked_add_u64(total: &mut u64, amount: u64, context: &str) -> Result<(), PersistenceError> {
    *total = total
        .checked_add(amount)
        .ok_or_else(|| PersistenceError::Invalid {
            context: format!("{context} overflows u64"),
        })?;
    Ok(())
}

/// Validate a lifecycle timeline and return exact cheap replay-work costs.
pub fn timeline_costs(
    events: &[SessionEvent],
    mode: Mode,
    final_tick: u64,
) -> Result<TimelineCosts, PersistenceError> {
    validate_events_inner(events, mode, final_tick, "events")
}

fn validate_events(
    events: &[SessionEvent],
    mode: Mode,
    final_tick: u64,
    context: &str,
) -> Result<TimelineCosts, PersistenceError> {
    validate_events_inner(events, mode, final_tick, context)
}

#[derive(Clone, Copy)]
struct EventView<'a> {
    base: &'a [SessionEvent],
    appended: Option<&'a SessionEvent>,
}

impl<'a> EventView<'a> {
    fn new(
        base: &'a [SessionEvent],
        appended: Option<&'a SessionEvent>,
    ) -> Result<Self, PersistenceError> {
        let view = Self { base, appended };
        view.checked_len()?;
        Ok(view)
    }

    fn checked_len(self) -> Result<usize, PersistenceError> {
        self.base
            .len()
            .checked_add(usize::from(self.appended.is_some()))
            .ok_or_else(|| PersistenceError::Limit {
                context: "event view length".into(),
                actual: usize::MAX,
                limit: MAX_SESSION_EVENTS,
            })
    }

    fn get(self, index: usize) -> Option<&'a SessionEvent> {
        self.base.get(index).or_else(|| {
            (index == self.base.len())
                .then_some(self.appended)
                .flatten()
        })
    }

    fn iter(self) -> impl Iterator<Item = &'a SessionEvent> {
        self.base.iter().chain(self.appended)
    }
}

fn validate_events_inner(
    events: &[SessionEvent],
    mode: Mode,
    final_tick: u64,
    context: &str,
) -> Result<TimelineCosts, PersistenceError> {
    validate_event_view(EventView::new(events, None)?, mode, final_tick, context)
}

fn validate_event_view(
    events: EventView<'_>,
    mode: Mode,
    final_tick: u64,
    context: &str,
) -> Result<TimelineCosts, PersistenceError> {
    limit(events.checked_len()?, MAX_SESSION_EVENTS, context)?;
    let mut cursor = 0u64;
    let mut costs = TimelineCosts::default();
    for (index, event) in events.iter().enumerate() {
        if event.pre_tick < cursor {
            return Err(PersistenceError::Invalid {
                context: format!("{context}[{index}].pre_tick precedes cursor {cursor}"),
            });
        }
        let idle = event.pre_tick - cursor;
        checked_add_u64(&mut costs.full_steps, idle, "full-step cost")?;
        checked_add_u64(&mut costs.total_elapsed_ticks, idle, "elapsed cost")?;
        match &event.kind {
            SessionEventKind::Action(action) => {
                let (full, unattended) = action_semantic_cost(action)?;
                let semantic =
                    full.checked_add(unattended)
                        .ok_or_else(|| PersistenceError::Invalid {
                            context: format!("{context}[{index}] semantic cost overflows"),
                        })?;
                let expected_post = event.pre_tick.checked_add(semantic).ok_or_else(|| {
                    PersistenceError::Invalid {
                        context: format!("{context}[{index}].post_tick overflows"),
                    }
                })?;
                if event.post_tick != expected_post {
                    return Err(PersistenceError::Invalid {
                        context: format!(
                            "{context}[{index}] action span is {}, expected {semantic}",
                            event.post_tick.saturating_sub(event.pre_tick)
                        ),
                    });
                }
                action_to_wire(action)?;
                checked_add_u64(&mut costs.full_steps, full, "full-step cost")?;
                checked_add_u64(&mut costs.unattended_steps, unattended, "unattended cost")?;
                checked_add_u64(&mut costs.total_elapsed_ticks, semantic, "elapsed cost")?;
            }
            SessionEventKind::StoryResurrect {
                checkpoint_after_event_count,
                ..
            } => {
                if mode != Mode::Story {
                    return Err(PersistenceError::Invalid {
                        context: format!("{context}[{index}] resurrection is Story-only"),
                    });
                }
                if event.post_tick > event.pre_tick {
                    return Err(PersistenceError::Invalid {
                        context: format!("{context}[{index}] resurrection advances time"),
                    });
                }
                validate_checkpoint_prefix(
                    events,
                    usize::try_from(*checkpoint_after_event_count).map_err(|_| {
                        PersistenceError::Invalid {
                            context: format!("{context}[{index}] checkpoint prefix is invalid"),
                        }
                    })?,
                    index,
                    event.post_tick,
                    &format!("{context}[{index}].checkpoint_after_event_count"),
                )?;
            }
        }
        cursor = event.post_tick;
    }
    if final_tick < cursor {
        return Err(PersistenceError::Invalid {
            context: format!("{context} final tick {final_tick} precedes cursor {cursor}"),
        });
    }
    let final_gap = final_tick - cursor;
    checked_add_u64(&mut costs.full_steps, final_gap, "full-step cost")?;
    checked_add_u64(&mut costs.total_elapsed_ticks, final_gap, "elapsed cost")?;
    Ok(costs)
}

fn validate_checkpoint_prefix(
    events: EventView<'_>,
    prefix: usize,
    upper_bound: usize,
    checkpoint_tick: u64,
    context: &str,
) -> Result<(), PersistenceError> {
    if prefix > upper_bound || prefix > events.checked_len()? {
        return Err(PersistenceError::Invalid {
            context: format!("{context} {prefix} is outside retained prefix {upper_bound}"),
        });
    }
    if prefix == 0 {
        if checkpoint_tick != 0 {
            return Err(PersistenceError::Invalid {
                context: format!("{context} zero prefix requires tick zero"),
            });
        }
        return Ok(());
    }
    let prior = events
        .get(prefix - 1)
        .ok_or_else(|| PersistenceError::Invalid {
            context: format!("{context} {prefix} is outside retained events"),
        })?;
    if !matches!(
        prior.kind,
        SessionEventKind::Action(CommittedAction::UseTransition { .. })
    ) || prior.post_tick != checkpoint_tick
    {
        return Err(PersistenceError::Invalid {
            context: format!("{context} must end on transition at checkpoint tick"),
        });
    }
    Ok(())
}

/// Encode exactly one pending-action descriptor as a tagged row. The slot
/// counters travel with the descriptor; the tag selects the typed payload.
fn pending_row_to_wire(pending: &Pending) -> Result<PendingRowWire, PersistenceError> {
    let mut wire = PendingRowWire {
        tag: pending.action.tag() as u8,
        start_tick: pending.start_tick,
        remaining_ticks: pending.remaining_ticks,
        direction: 0,
        service: 0,
        text: BoundedString::default(),
        id: 0,
        inn_total: 0,
        inn_remaining: 0,
    };
    let bad = |detail: &str| PersistenceError::Invalid {
        context: format!("state.pending {detail}"),
    };
    match &pending.action {
        PendingAction::Move { direction } => wire.direction = direction.canonical_tag(),
        PendingAction::Wait => {}
        PendingAction::InspectBoard { town } => {
            valid_string(town, "state.pending.town", true)?;
            wire.text = town.clone().into();
        }
        PendingAction::AcceptContract { id } => {
            valid_string(id, "state.pending.accept_contract.id", true)?;
            wire.text = id.clone().into();
        }
        PendingAction::UseTransition { transition_id } => {
            if *transition_id == 0 {
                return Err(bad("transition id is zero"));
            }
            wire.id = *transition_id;
        }
        PendingAction::ResolveAttack { target_actor } => {
            if *target_actor == 0 {
                return Err(bad("target actor is zero"));
            }
            wire.id = *target_actor;
        }
        PendingAction::Pickup { item_id }
        | PendingAction::Drop { item_id }
        | PendingAction::Wield { item_id }
        | PendingAction::TinkerIdentify { item_id } => {
            if *item_id == 0 {
                return Err(bad("item id is zero"));
            }
            wire.id = *item_id;
        }
        PendingAction::TurnIn { contract_id } => {
            valid_string(contract_id, "state.pending.turn_in.contract_id", true)?;
            wire.text = contract_id.clone().into();
        }
        PendingAction::UseService { service } => wire.service = service.canonical_tag(),
        PendingAction::Inn(inn) => {
            wire.inn_total = inn.total_ticks;
            wire.inn_remaining = inn.remaining_ticks;
        }
    }
    Ok(wire)
}

fn action_to_wire(action: &CommittedAction) -> Result<ActionWire, PersistenceError> {
    let mut wire = ActionWire {
        tag: 0,
        direction: 0,
        service: 0,
        text: BoundedString::default(),
        id: 0,
    };
    match action {
        CommittedAction::Move(direction) => {
            wire.tag = 1;
            wire.direction = direction_tag(direction);
        }
        CommittedAction::Wait => wire.tag = 2,
        CommittedAction::InspectBoard { town } => {
            valid_string(town, "action.inspect_board.town", true)?;
            wire.tag = 3;
            wire.text = town.clone().into();
        }
        CommittedAction::AcceptContract { id } => {
            valid_string(id, "action.accept_contract.id", true)?;
            wire.tag = 4;
            wire.text = id.clone().into();
        }
        CommittedAction::UseTransition { transition_id } => {
            if *transition_id == 0 {
                return Err(PersistenceError::Invalid {
                    context: "action transition id is zero".into(),
                });
            }
            wire.tag = 5;
            wire.id = *transition_id;
        }
        CommittedAction::ResolveAttack { target_actor } => {
            if *target_actor == 0 {
                return Err(PersistenceError::Invalid {
                    context: "action target actor is zero".into(),
                });
            }
            wire.tag = 6;
            wire.id = *target_actor;
        }
        CommittedAction::Pickup { item_id } => {
            if *item_id == 0 {
                return Err(PersistenceError::Invalid {
                    context: "action pickup item id is zero".into(),
                });
            }
            wire.tag = 9;
            wire.id = *item_id;
        }
        CommittedAction::Drop { item_id } => {
            if *item_id == 0 {
                return Err(PersistenceError::Invalid {
                    context: "action drop item id is zero".into(),
                });
            }
            wire.tag = 10;
            wire.id = *item_id;
        }
        // Additive v4 tag: current readers retain old-record compatibility;
        // older readers reject Wield-bearing records as an unknown action tag.
        CommittedAction::Wield { item_id } => {
            if *item_id == 0 {
                return Err(PersistenceError::Invalid {
                    context: "action wield item id is zero".into(),
                });
            }
            wire.tag = 11;
            wire.id = *item_id;
        }
        // Additive canonical v4 tag; tags 1–11 retain their original payloads.
        CommittedAction::TinkerIdentify { item_id } => {
            if *item_id == 0 {
                return Err(PersistenceError::Invalid {
                    context: "action tinker identify item id is zero".into(),
                });
            }
            wire.tag = 12;
            wire.id = *item_id;
        }
        CommittedAction::TurnIn { contract_id } => {
            valid_string(contract_id, "action.turn_in.contract_id", true)?;
            wire.tag = 7;
            wire.text = contract_id.clone().into();
        }
        CommittedAction::UseService { service } => {
            wire.tag = 8;
            wire.service = service_tag(*service);
        }
    }
    Ok(wire)
}

fn wire_to_action(wire: ActionWire, context: &str) -> Result<CommittedAction, PersistenceError> {
    let unused = |direction, service, text: &str, id| {
        if direction == 0 && service == 0 && text.is_empty() && id == 0 {
            Ok(())
        } else {
            Err(PersistenceError::Invalid {
                context: format!("{context} has noncanonical unused payload"),
            })
        }
    };
    match wire.tag {
        1 => {
            if wire.service != 0 || !wire.text.is_empty() || wire.id != 0 {
                return Err(PersistenceError::Invalid {
                    context: format!("{context} move has unused payload"),
                });
            }
            Ok(CommittedAction::Move(direction(
                wire.direction,
                &format!("{context}.direction"),
            )?))
        }
        2 => {
            unused(wire.direction, wire.service, &wire.text, wire.id)?;
            Ok(CommittedAction::Wait)
        }
        3 | 4 | 7 => {
            if wire.direction != 0 || wire.service != 0 || wire.id != 0 {
                return Err(PersistenceError::Invalid {
                    context: format!("{context} text action has unused payload"),
                });
            }
            valid_string(&wire.text, &format!("{context}.text"), true)?;
            let text = wire.text.into_inner();
            Ok(match wire.tag {
                3 => CommittedAction::InspectBoard { town: text },
                4 => CommittedAction::AcceptContract { id: text },
                _ => CommittedAction::TurnIn { contract_id: text },
            })
        }
        5 | 6 | 9 | 10 | 11 | 12 => {
            if wire.direction != 0 || wire.service != 0 || !wire.text.is_empty() || wire.id == 0 {
                return Err(PersistenceError::Invalid {
                    context: format!("{context} id action payload is noncanonical"),
                });
            }
            Ok(match wire.tag {
                5 => CommittedAction::UseTransition {
                    transition_id: wire.id,
                },
                6 => CommittedAction::ResolveAttack {
                    target_actor: wire.id,
                },
                9 => CommittedAction::Pickup { item_id: wire.id },
                10 => CommittedAction::Drop { item_id: wire.id },
                11 => CommittedAction::Wield { item_id: wire.id },
                12 => CommittedAction::TinkerIdentify { item_id: wire.id },
                _ => unreachable!("matched only canonical id-action tags"),
            })
        }
        8 => {
            if wire.direction != 0 || !wire.text.is_empty() || wire.id != 0 {
                return Err(PersistenceError::Invalid {
                    context: format!("{context} service has unused payload"),
                });
            }
            Ok(CommittedAction::UseService {
                service: service(wire.service, &format!("{context}.service"))?,
            })
        }
        tag => Err(PersistenceError::Invalid {
            context: format!("{context}.tag unknown {tag}"),
        }),
    }
}

/// Inverse of [`pending_row_to_wire`]: decode exactly one pending row and
/// validate its tag-selected payload (checked decoding; unused fields must be
/// canonical zero/defaults).
fn pending_row_from_wire(wire: PendingRowWire, context: &str) -> Result<Pending, PersistenceError> {
    let bad = |detail: &str| PersistenceError::Invalid {
        context: format!("{context} {detail}"),
    };
    let tag = PendingTag::from_tag(wire.tag)
        .ok_or_else(|| bad(&format!("unknown tag {}", wire.tag)))?;
    let text_clear = |wire: &PendingRowWire| {
        wire.direction == 0
            && wire.service == 0
            && wire.id == 0
            && wire.inn_total == 0
            && wire.inn_remaining == 0
    };
    let id_clear = |wire: &PendingRowWire| {
        wire.direction == 0
            && wire.service == 0
            && wire.text.is_empty()
            && wire.inn_total == 0
            && wire.inn_remaining == 0
    };
    let action = match tag {
        PendingTag::Move => {
            if wire.service != 0
                || !wire.text.is_empty()
                || wire.id != 0
                || wire.inn_total != 0
                || wire.inn_remaining != 0
                || wire.direction == 0
            {
                return Err(bad("move payload is noncanonical"));
            }
            PendingAction::Move {
                direction: direction(wire.direction, &format!("{context}.direction"))?,
            }
        }
        PendingTag::Wait => {
            if !pending_payload_clear(&wire) {
                return Err(bad("wait has unused payload"));
            }
            PendingAction::Wait
        }
        PendingTag::InspectBoard | PendingTag::AcceptContract | PendingTag::TurnIn => {
            if !text_clear(&wire) {
                return Err(bad("text action has unused payload"));
            }
            valid_string(&wire.text, &format!("{context}.text"), true)?;
            let text = wire.text.clone().into_inner();
            match tag {
                PendingTag::InspectBoard => PendingAction::InspectBoard { town: text },
                PendingTag::AcceptContract => PendingAction::AcceptContract { id: text },
                _ => PendingAction::TurnIn { contract_id: text },
            }
        }
        PendingTag::UseTransition
        | PendingTag::ResolveAttack
        | PendingTag::Pickup
        | PendingTag::Drop
        | PendingTag::Wield
        | PendingTag::TinkerIdentify => {
            if !id_clear(&wire) || wire.id == 0 {
                return Err(bad("id action payload is noncanonical"));
            }
            match tag {
                PendingTag::UseTransition => PendingAction::UseTransition {
                    transition_id: wire.id,
                },
                PendingTag::ResolveAttack => PendingAction::ResolveAttack {
                    target_actor: wire.id,
                },
                PendingTag::Pickup => PendingAction::Pickup { item_id: wire.id },
                PendingTag::Drop => PendingAction::Drop { item_id: wire.id },
                PendingTag::Wield => PendingAction::Wield { item_id: wire.id },
                _ => PendingAction::TinkerIdentify { item_id: wire.id },
            }
        }
        PendingTag::UseService => {
            if wire.direction != 0
                || !wire.text.is_empty()
                || wire.id != 0
                || wire.inn_total != 0
                || wire.inn_remaining != 0
            {
                return Err(bad("service has unused payload"));
            }
            PendingAction::UseService {
                service: service(wire.service, &format!("{context}.service"))?,
            }
        }
        PendingTag::Inn => {
            if wire.direction != 0 || wire.service != 0 || !wire.text.is_empty() || wire.id != 0 {
                return Err(bad("inn has unused payload"));
            }
            PendingAction::Inn(PendingInn {
                total_ticks: wire.inn_total,
                remaining_ticks: wire.inn_remaining,
            })
        }
    };
    Ok(Pending {
        start_tick: wire.start_tick,
        remaining_ticks: wire.remaining_ticks,
        action,
    })
}

fn pending_payload_clear(wire: &PendingRowWire) -> bool {
    wire.direction == 0
        && wire.service == 0
        && wire.text.is_empty()
        && wire.id == 0
        && wire.inn_total == 0
        && wire.inn_remaining == 0
}

/// The canonical text payload of a pending operation, if it carries one.
fn pending_text(pending: &Pending) -> Option<&str> {
    match &pending.action {
        PendingAction::InspectBoard { town } => Some(town),
        PendingAction::AcceptContract { id } => Some(id),
        PendingAction::TurnIn { contract_id } => Some(contract_id),
        _ => None,
    }
}

/// Encode one observation row with checked provenance. The location key must be
/// a valid Town or Deep key (never the reserved uninitialized location) and the
/// face tag must be canonical.
fn observation_row_to_wire(row: &ObservationRow) -> Result<ObservationRowWire, PersistenceError> {
    let bad = |detail: String| PersistenceError::Invalid {
        context: format!("state.observations {detail}"),
    };
    let (location_kind, town, depth) = match row.location.to_run_location() {
        Some(RunLocation::Town(town)) => {
            valid_string(&town, "state.observations.town", true)?;
            (LOCATION_KIND_TOWN, town.into(), 0)
        }
        Some(RunLocation::Deep(depth)) => (LOCATION_KIND_DEEP, BoundedString::default(), depth),
        _ => return Err(bad("location is not a live Town/Deep key".into())),
    };
    if Face::from_tag(row.face).is_none() {
        return Err(bad(format!("unknown face tag {}", row.face)));
    }
    if row.appearance.door_appearance > DOOR_APPEARANCE_GLASS {
        return Err(bad(format!(
            "unknown door appearance {}",
            row.appearance.door_appearance
        )));
    }
    Ok(ObservationRowWire {
        location_kind,
        town,
        depth,
        cell: cell(row.cell),
        face: row.face,
        material: row.appearance.material,
        occupancy: row.appearance.occupancy,
        door_appearance: row.appearance.door_appearance,
        observed_tick: row.appearance.observed_tick,
        branch_epoch: row.appearance.branch_epoch,
    })
}

/// Inverse of [`observation_row_to_wire`] with checked decoding: the location
/// kind/depth/town combination and the face tag are validated, unknown tags are
/// rejected rather than defaulted.
fn observation_row_from_wire(
    wire: ObservationRowWire,
    context: &str,
) -> Result<ObservationRow, PersistenceError> {
    let bad = |detail: String| PersistenceError::Invalid {
        context: format!("{context} {detail}"),
    };
    let location = match wire.location_kind {
        LOCATION_KIND_TOWN => {
            if wire.depth != 0 {
                return Err(bad("town row carries a depth".into()));
            }
            valid_string(&wire.town, &format!("{context}.town"), true)?;
            LocationKey {
                kind: LOCATION_KIND_TOWN,
                town: wire.town.into_inner(),
                depth: 0,
            }
        }
        LOCATION_KIND_DEEP => {
            if !(1..=12).contains(&wire.depth) {
                return Err(bad(format!("deep row depth {} out of range", wire.depth)));
            }
            if !wire.town.is_empty() {
                return Err(bad("deep row carries a town".into()));
            }
            LocationKey {
                kind: LOCATION_KIND_DEEP,
                town: String::new(),
                depth: wire.depth,
            }
        }
        other => return Err(bad(format!("unknown location kind {other}"))),
    };
    let face = Face::from_tag(wire.face)
        .ok_or_else(|| bad(format!("unknown face tag {}", wire.face)))?
        .tag();
    if wire.door_appearance > DOOR_APPEARANCE_GLASS {
        return Err(bad(format!(
            "unknown door appearance {}",
            wire.door_appearance
        )));
    }
    Ok(ObservationRow {
        location,
        cell: uncell(wire.cell),
        face,
        appearance: ObservedFace {
            material: wire.material,
            occupancy: wire.occupancy,
            door_appearance: wire.door_appearance,
            observed_tick: wire.observed_tick,
            branch_epoch: wire.branch_epoch,
        },
    })
}

/// Reject a duplicate or non-canonically ordered observation row vector. Rows
/// are ordered by (location, cell, face).
fn require_strict_observation_order(rows: &[ObservationRow]) -> Result<(), PersistenceError> {
    let mut previous: Option<(&LocationKey, (i32, i32, i32), u8)> = None;
    for (index, row) in rows.iter().enumerate() {
        let key = (&row.location, row.cell, row.face);
        if let Some(prior) = previous {
            let ordered =
                key.0 > prior.0 || (key.0 == prior.0 && (key.1 > prior.1 || (key.1 == prior.1 && key.2 > prior.2)));
            if !ordered {
                return Err(PersistenceError::Invalid {
                    context: format!("state.observations[{index}] is duplicate or noncanonical order"),
                });
            }
        }
        previous = Some(key);
    }
    Ok(())
}

fn event_to_wire(event: &SessionEvent) -> Result<SessionEventWire, PersistenceError> {
    let mut wire = SessionEventWire {
        pre_tick: event.pre_tick,
        post_tick: event.post_tick,
        kind_tag: 0,
        action: ActionWire {
            tag: 0,
            direction: 0,
            service: 0,
            text: BoundedString::default(),
            id: 0,
        },
        checkpoint_after_event_count: 0,
        checkpoint_hash: 0,
        dead_state_hash: 0,
        restored_state_hash: 0,
    };
    match &event.kind {
        SessionEventKind::Action(action) => {
            wire.kind_tag = 1;
            wire.action = action_to_wire(action)?;
        }
        SessionEventKind::StoryResurrect {
            checkpoint_after_event_count,
            checkpoint_hash,
            dead_state_hash,
            restored_state_hash,
        } => {
            wire.kind_tag = 2;
            wire.checkpoint_after_event_count = *checkpoint_after_event_count;
            wire.checkpoint_hash = *checkpoint_hash;
            wire.dead_state_hash = *dead_state_hash;
            wire.restored_state_hash = *restored_state_hash;
        }
    }
    Ok(wire)
}

fn wire_to_event(
    wire: SessionEventWire,
    index: usize,
    context: &str,
) -> Result<SessionEvent, PersistenceError> {
    let kind = match wire.kind_tag {
        1 => {
            if wire.checkpoint_after_event_count != 0
                || wire.checkpoint_hash != 0
                || wire.dead_state_hash != 0
                || wire.restored_state_hash != 0
            {
                return Err(PersistenceError::Invalid {
                    context: format!("{context}[{index}] action has lifecycle payload"),
                });
            }
            SessionEventKind::Action(wire_to_action(
                wire.action,
                &format!("{context}[{index}].action"),
            )?)
        }
        2 => {
            if wire.action.tag != 0
                || wire.action.direction != 0
                || wire.action.service != 0
                || !wire.action.text.is_empty()
                || wire.action.id != 0
            {
                return Err(PersistenceError::Invalid {
                    context: format!("{context}[{index}] resurrection has action payload"),
                });
            }
            SessionEventKind::StoryResurrect {
                checkpoint_after_event_count: wire.checkpoint_after_event_count,
                checkpoint_hash: wire.checkpoint_hash,
                dead_state_hash: wire.dead_state_hash,
                restored_state_hash: wire.restored_state_hash,
            }
        }
        tag => {
            return Err(PersistenceError::Invalid {
                context: format!("{context}[{index}].kind_tag unknown {tag}"),
            })
        }
    };
    Ok(SessionEvent {
        pre_tick: wire.pre_tick,
        post_tick: wire.post_tick,
        kind,
    })
}

fn wires_to_events(
    rows: Vec<SessionEventWire>,
    context: &str,
) -> Result<Vec<SessionEvent>, PersistenceError> {
    limit(rows.len(), MAX_SESSION_EVENTS, context)?;
    rows.into_iter()
        .enumerate()
        .map(|(index, wire)| wire_to_event(wire, index, context))
        .collect()
}

/// SHA-256 over an explicit canonical wire encoding of the requested event
/// prefix. Events appended after `count` do not affect the digest.
pub fn event_prefix_digest(
    events: &[SessionEvent],
    count: usize,
) -> Result<[u8; 32], PersistenceError> {
    limit(events.len(), MAX_SESSION_EVENTS, "events")?;
    if count > events.len() {
        return Err(PersistenceError::Invalid {
            context: format!("event digest prefix {count} exceeds {}", events.len()),
        });
    }
    let mut budget = EncodeBudget::default();
    budget.check_events(&events[..count], "event_digest.events")?;
    let mut hasher = Sha256::new();
    hasher.update(b"hexwork-session-event-prefix-v3\0");
    hasher.update((count as u64).to_le_bytes());
    for event in &events[..count] {
        let wire = event_to_wire(event)?;
        let encoded = bincode::encode_to_vec(wire, bincode::config::standard())
            .map_err(|e| PersistenceError::Decode(e.to_string()))?;
        hasher.update((encoded.len() as u64).to_le_bytes());
        hasher.update(encoded);
    }
    Ok(hasher.finalize().into())
}

/// Validate structural metadata, canonical prospective history, and all logical
/// run count/string budgets.
///
/// The caller must already have validated the current [`StateImage`] through
/// core/runtime invariants. This helper deliberately does not reconstruct or
/// hash either image, prove retained history, serialize, compress, or perform
/// I/O. Success is structural admission only: later serialization, raw or
/// compressed byte caps, and durable I/O can still fail, and success does not
/// guarantee that this run or any future state can be saved.
pub fn validate_run_structural_limits(
    identity: &RunIdentity,
    current: &StateImage,
    story_checkpoint: Option<&StoryCheckpoint>,
    committed_events: &[SessionEvent],
    appended_event: Option<&SessionEvent>,
) -> Result<(), PersistenceError> {
    let events = EventView::new(committed_events, appended_event)?;
    validate_identity(identity, "run.identity")?;
    metadata(identity.seed, current.seed, "seed")?;
    validate_event_view(events, identity.mode, current.tick, "run.events")?;
    let current_template = validate_player_image(current, identity, false, "run.current")?;
    match identity.mode {
        Mode::Story => {
            let checkpoint = story_checkpoint.ok_or_else(|| PersistenceError::Invalid {
                context: "Story run requires exactly one active checkpoint".into(),
            })?;
            metadata(identity.seed, checkpoint.state.image.seed, "seed")?;
            let checkpoint_template = validate_player_image(
                &checkpoint.state.image,
                identity,
                true,
                "run.story_checkpoint",
            )?;
            if checkpoint_template != current_template {
                return Err(PersistenceError::Invalid {
                    context: "current and checkpoint Player templates differ".into(),
                });
            }
            let prefix = usize::try_from(checkpoint.after_event_count).map_err(|_| {
                PersistenceError::Invalid {
                    context: "checkpoint event prefix is invalid".into(),
                }
            })?;
            validate_checkpoint_prefix(
                events,
                prefix,
                events.checked_len()?,
                checkpoint.state.tick,
                "run.story_checkpoint.after_event_count",
            )?;
        }
        Mode::Classic | Mode::Ironcartographer => {
            if story_checkpoint.is_some() {
                return Err(PersistenceError::Invalid {
                    context: "Classic/Iron run cannot contain a Story checkpoint".into(),
                });
            }
        }
    }
    let mut budget = EncodeBudget::default();
    budget.check_identity(identity, "run.identity")?;
    budget.check_event_view(events, "run.events")?;
    budget.check_image(current)?;
    if let Some(checkpoint) = story_checkpoint {
        budget.check_image(&checkpoint.state.image)?;
    }
    Ok(())
}

fn validate_player_image<'a>(
    image: &'a StateImage,
    identity: &RunIdentity,
    checkpoint: bool,
    context: &str,
) -> Result<&'a str, PersistenceError> {
    if checkpoint && image.run_location != RunLocation::Town("milthollow".into()) {
        return Err(PersistenceError::Invalid {
            context: format!("{context} must be canonical Town milthollow"),
        });
    }
    let players: Vec<_> = image
        .actor_identities
        .iter()
        .filter(|(_, actor_identity)| actor_identity.role == ActorRole::Player)
        .collect();
    if players.len() != 1 {
        return Err(PersistenceError::Invalid {
            context: format!("{context} must contain exactly one Player"),
        });
    }
    let (player_id, actor_identity) = players[0];
    if actor_identity.display_name != identity.display_name {
        return Err(PersistenceError::Invalid {
            context: format!("{context} Player display name differs from run identity"),
        });
    }
    let actor = image
        .actors
        .iter()
        .find(|actor| actor.id == *player_id)
        .ok_or_else(|| PersistenceError::Invalid {
            context: format!("{context} Player identity has no actor"),
        })?;
    let character = image
        .characters
        .get(player_id)
        .ok_or_else(|| PersistenceError::Invalid {
            context: format!("{context} Player has no Character"),
        })?;
    if character.ancestry != identity.ancestry || character.calling != identity.calling {
        return Err(PersistenceError::Invalid {
            context: format!("{context} Player ancestry/calling differs from run identity"),
        });
    }
    if checkpoint && (!actor.alive || actor.hp <= 0) {
        return Err(PersistenceError::Invalid {
            context: format!("{context} Player must be living with positive HP"),
        });
    }
    Ok(&actor_identity.template_id)
}

fn checked_value(
    current: usize,
    amount: usize,
    cap: usize,
    context: &str,
) -> Result<usize, PersistenceError> {
    let mut value = current;
    checked_total(&mut value, amount, cap, context)?;
    Ok(value)
}

fn checked_total(
    total: &mut usize,
    amount: usize,
    cap: usize,
    context: &str,
) -> Result<(), PersistenceError> {
    *total = total
        .checked_add(amount)
        .ok_or_else(|| PersistenceError::Limit {
            context: context.into(),
            actual: usize::MAX,
            limit: cap,
        })?;
    limit(*total, cap, context)
}

fn validate_image_limits(i: &StateImage) -> Result<(), PersistenceError> {
    limit(i.chunks.len(), MAX_CHUNKS, "state.chunks")?;
    limit(
        i.observations.len(),
        MAX_OBSERVATION_ROWS,
        "state.observations",
    )?;
    let mut state_rows = 0;
    for (name, rows) in [
        ("chunks", i.chunks.len()),
        ("actors", i.actors.len()),
        ("identities", i.actor_identities.len()),
        ("wallets", i.wallets.len()),
        ("contracts", i.contracts.len()),
        ("connectors", i.connectors.len()),
        ("pressure_nodes", i.pressure_nodes.len()),
        ("pressure_edges", i.pressure_edges.len()),
        ("causes", i.causes.len()),
        ("laws", i.laws.len()),
        ("fall_odo", i.fall_odo.len()),
        ("spill.desired_moves", i.spill.desired_moves.len()),
        ("spill.crossed_cells", i.spill.crossed_cells.len()),
        ("lights", i.lights.len()),
        ("sounds", i.sounds.len()),
        ("corpses", i.corpses.len()),
        ("signal_sources", i.signal_sources.len()),
        ("characters", i.characters.len()),
        ("reputations", i.reputations.len()),
        ("items", i.items.len()),
        ("ground_items", i.ground_items.len()),
        ("packs", i.packs.len()),
        ("idf", i.idf.len()),
        ("hymn_cases", i.hymn_cases.len()),
        ("cast_queue", i.cast_queue.len()),
        ("bargains", i.bargains.len()),
        ("npcs", i.npcs.len()),
        ("shops", i.shops.len()),
        ("gossip", i.gossip.len()),
        ("chalk", i.chalk.marks.len()),
        ("chapel_debts", i.chapel_debts.len()),
        ("observations", i.observations.len()),
    ] {
        limit(rows, MAX_ROWS, &format!("state.{name}"))?;
        checked_total(
            &mut state_rows,
            rows,
            MAX_TOTAL_STATE_ROWS,
            "total state rows",
        )?;
    }

    let mut nested_rows = 0;
    for (index, chunk) in i.chunks.iter().enumerate() {
        limit(
            chunk.voxels.len(),
            CHUNK_VOL,
            &format!("state.chunks[{index}].voxels"),
        )?;
        checked_total(
            &mut nested_rows,
            chunk.voxels.len(),
            MAX_TOTAL_NESTED_ROWS,
            "total nested rows",
        )?;
    }
    let gas_flows = i.spill.gas_flows.len();
    limit(gas_flows, MAX_ROWS, "state.spill.gas_flows")?;
    limit(gas_flows, MAX_TOTAL_GAS_FLOWS, "spill gas flows")?;
    checked_total(
        &mut nested_rows,
        gas_flows,
        MAX_TOTAL_NESTED_ROWS,
        "total nested rows",
    )?;
    let mule_waypoints = i.mule.as_ref().map_or(0, |mule| mule.waypoints.len());
    limit(mule_waypoints, MAX_TOTAL_MULE_WAYPOINTS, "mule waypoints")?;
    checked_total(
        &mut nested_rows,
        mule_waypoints,
        MAX_TOTAL_NESTED_ROWS,
        "total nested rows",
    )?;

    let mut town_rows = 0;
    let mut pack_items = 0;
    let mut hymn_cards = 0;
    let mut hymn_miscasts = 0;
    let mut ledger_debts = 0;
    let mut shop_rows = 0;
    let mut item_tags = 0;
    for reputation in i.reputations.values() {
        limit(
            reputation.town_ledger.len(),
            MAX_TOWN_LEDGER_ROWS,
            "reputation.town_ledger",
        )?;
        checked_total(
            &mut town_rows,
            reputation.town_ledger.len(),
            MAX_TOTAL_TOWN_LEDGER_ROWS,
            "reputation town-ledger rows",
        )?;
    }
    for item in i.items.values() {
        limit(item.tags.len(), MAX_ITEM_TAGS, "item.tags")?;
        checked_total(
            &mut item_tags,
            item.tags.len(),
            MAX_TOTAL_ITEM_TAGS,
            "item tags",
        )?;
    }
    for pack in i.packs.values() {
        limit(pack.items.len(), MAX_PACK_ITEMS, "pack.items")?;
        checked_total(
            &mut pack_items,
            pack.items.len(),
            MAX_TOTAL_PACK_ITEMS,
            "pack items",
        )?;
        for item in pack.items.values() {
            limit(item.tags.len(), MAX_ITEM_TAGS, "pack.item.tags")?;
            checked_total(
                &mut item_tags,
                item.tags.len(),
                MAX_TOTAL_ITEM_TAGS,
                "item tags",
            )?;
        }
    }
    for case in i.hymn_cases.values() {
        limit(case.cards.len(), MAX_HYMN_CARDS, "hymn_case.cards")?;
        checked_total(
            &mut hymn_cards,
            case.cards.len(),
            MAX_TOTAL_HYMN_CARDS,
            "hymn cards",
        )?;
        for card in &case.cards {
            limit(card.miscasts.len(), MAX_HYMN_MISCASTS, "hymn_card.miscasts")?;
            checked_total(
                &mut hymn_miscasts,
                card.miscasts.len(),
                MAX_TOTAL_HYMN_MISCASTS,
                "hymn miscasts",
            )?;
        }
    }
    for ledger in i.bargains.values() {
        limit(ledger.debts.len(), MAX_LEDGER_DEBTS, "ledger.debts")?;
        checked_total(
            &mut ledger_debts,
            ledger.debts.len(),
            MAX_TOTAL_LEDGER_DEBTS,
            "ledger debts",
        )?;
    }
    for shop in i.shops.values() {
        for (name, rows) in [
            ("shop.broke", shop.broke.len()),
            ("shop.stole", shop.stole.len()),
            ("shop.killed", shop.killed.len()),
        ] {
            limit(rows, MAX_SHOP_ROWS, name)?;
            checked_total(
                &mut shop_rows,
                rows,
                MAX_TOTAL_SHOP_ROWS,
                "shop memory rows",
            )?;
        }
    }
    for (amount, cap, context) in [
        (
            town_rows,
            MAX_TOTAL_TOWN_LEDGER_ROWS,
            "reputation town-ledger rows",
        ),
        (pack_items, MAX_TOTAL_PACK_ITEMS, "pack items"),
        (hymn_cards, MAX_TOTAL_HYMN_CARDS, "hymn cards"),
        (hymn_miscasts, MAX_TOTAL_HYMN_MISCASTS, "hymn miscasts"),
        (ledger_debts, MAX_TOTAL_LEDGER_DEBTS, "ledger debts"),
        (shop_rows, MAX_TOTAL_SHOP_ROWS, "shop memory rows"),
        (item_tags, MAX_TOTAL_ITEM_TAGS, "item tags"),
    ] {
        limit(amount, cap, context)?;
        checked_total(
            &mut nested_rows,
            amount,
            MAX_TOTAL_NESTED_ROWS,
            "total nested rows",
        )?;
    }
    limit(i.pending.len(), MAX_PENDING_ROWS, "state.pending")?;
    let mut strings = EncodeBudget::default();
    strings.check_image_strings(i)
}

fn image_to_wire(i: StateImage) -> Result<StateWire, PersistenceError> {
    validate_image_limits(&i)?;
    Ok(StateWire {
        tick: i.tick,
        seed: i.seed,
        rng_master: i.rng_master,
        ecs_entity_count: i.ecs_entity_count,
        chunks: i
            .chunks
            .into_iter()
            .map(|c| ChunkWire {
                coord: cell(c.coord),
                voxels: c.voxels.into_iter().map(voxel_to_wire).collect(),
            })
            .collect(),
        actors: i
            .actors
            .into_iter()
            .map(|a| ActorWire {
                id: a.id,
                position_mm: a.position.millimetres(),
                body_mm: a.body.millimetres(),
                heading: a.heading.map(|h| h.bin()),
                movement_remainder: a.movement_remainder.numerators(),
                stance: a.stance.canonical_tag(),
                hp: a.hp,
                alive: a.alive,
            })
            .collect(),
        identities: i
            .actor_identities
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    IdentityWire {
                        template: v.template_id.into(),
                        name: v.display_name.into(),
                        role: v.role.canonical_tag(),
                    },
                )
            })
            .collect(),
        wallets: i
            .wallets
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    WalletWire {
                        copper: v.copper(),
                        scrip: v.scrip(),
                    },
                )
            })
            .collect(),
        contracts: i
            .contracts
            .into_iter()
            .map(|(k, v)| {
                (
                    k.into(),
                    ContractWire {
                        id: v.id.into(),
                        taker: v.taker,
                        accepted_tick: v.accepted_tick,
                        floor: v.floor,
                        pay_cu: v.pay_cu,
                        target: v.objective_target,
                        progress: v.objective_progress,
                        status: v.status.canonical_tag(),
                        reward_town: v.reward_town.into(),
                        standing_delta: v.standing_delta,
                    },
                )
            })
            .collect(),
        location: location_to_wire(i.run_location),
        connectors: i
            .connectors
            .into_iter()
            .map(|v| ConnectorWire {
                id: v.id,
                kind: v.kind.canonical_tag(),
                from: v.from.map(f32::to_bits),
                to: v.to.map(f32::to_bits),
                length: v.length.to_bits(),
                anchor_quality: v.anchor_quality,
                load_kg: v.load_kg.to_bits(),
                blocked: v.blocked,
            })
            .collect(),
        pressure_nodes: i
            .pressure_nodes
            .into_iter()
            .map(|v| PressureNodeWire {
                id: v.id,
                pressure: v.pressure.to_bits(),
            })
            .collect(),
        pressure_edges: i
            .pressure_edges
            .into_iter()
            .map(|v| PressureEdgeWire {
                from: v.from,
                to: v.to,
                openness: v.openness.to_bits(),
                hidden: v.hidden,
            })
            .collect(),
        causes: i
            .causes
            .into_iter()
            .map(|v| CauseWire {
                tick: v.tick,
                rule: v.rule.into(),
                detail: v.detail.into(),
            })
            .collect(),
        far: FarWire {
            day: i.far.day,
            day_tick: i.far.day_tick,
        },
        laws: i
            .laws
            .into_iter()
            .map(|v| LawWire {
                rule: v.rule.into(),
                tick: v.first_seen_tick,
            })
            .collect(),
        fall_odo: i.fall_odo.into_iter().collect(),
        lights: i
            .lights
            .into_iter()
            .map(|v| LightWire {
                x: v.x,
                y: v.y,
                z: v.z,
                radius: v.radius,
                warm: v.warm,
            })
            .collect(),
        sounds: i
            .sounds
            .into_iter()
            .map(|v| SoundWire {
                x: v.x,
                y: v.y,
                z: v.z,
                loudness: v.loudness,
                timbre: v.timbre,
            })
            .collect(),
        corpses: i
            .corpses
            .into_iter()
            .map(|(k, v)| {
                (
                    cell(k),
                    CorpseWire {
                        age_ticks: v.age_ticks,
                        bursts_at: v.bursts_at,
                    },
                )
            })
            .collect(),
        signal_sources: i.signal_sources.into_iter().map(cell).collect(),
        spill: SpillWire {
            desired_moves: i.spill.desired_moves.into_iter().map(|(id,s)| (id,s.heading.map(|h|h.bin()),s.speed_mm_s,s.stop_at.map(|p|p.millimetres()))).collect(),
            crossed_cells: i.spill.crossed_cells.into_iter().map(|(id,cells)| (id,cells.into_iter().map(cell).collect())).collect(),
            gas_flows: i
                .spill
                .gas_flows
                .into_iter()
                .map(|(a, b, c)| (a, b, c.to_bits()))
                .collect(),
            light_pending: i.spill.light_pending,
            signal_pending: i.spill.signal_pending,
            scent_pending: i.spill.scent_pending,
        },
        characters: i
            .characters
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    CharacterWire {
                        ancestry: v.ancestry.canonical_tag(),
                        calling: v.calling.canonical_tag(),
                        attrs: v.attrs,
                        skills: v.skills,
                        uses: v.uses,
                        pips: v.pips,
                    },
                )
            })
            .collect(),
        reputations: i
            .reputations
            .into_iter()
            .map(|(k, v)| (k, reputation_to_wire(v)))
            .collect(),
        items: i
            .items
            .into_iter()
            .map(|(k, v)| (k, item_to_wire(v)))
            .collect(),
        ground_items: i
            .ground_items
            .into_iter()
            .map(|(k, v)| (k, cell(v)))
            .collect(),
        packs: i
            .packs
            .into_iter()
            .map(|(k, v)| (k, pack_to_wire(v)))
            .collect(),
        knowledge: knowledge_to_wire(i.knowledge),
        idf: i.idf.into_iter().map(|(k, v)| (k.into(), v)).collect(),
        hymn_cases: i
            .hymn_cases
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    HymnCaseWire {
                        cards: v
                            .cards
                            .into_iter()
                            .map(|c| HymnCardWire {
                                template: c.template.into(),
                                conduct_base: c.conduct_base,
                                slots: c.slots,
                                fragile: c.fragile,
                                miscasts: c.miscasts.into_iter().map(Into::into).collect(),
                            })
                            .collect(),
                        max_slots: v.max_slots,
                    },
                )
            })
            .collect(),
        cast_queue: i
            .cast_queue
            .into_iter()
            .map(|(caster, template, target)| CastWire {
                caster,
                template: template.into(),
                target: cell((target[0], target[1], target[2])),
            })
            .collect(),
        bargains: i
            .bargains
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    LedgerWire {
                        debts: v
                            .debts
                            .into_iter()
                            .map(|d| DebtWire {
                                god: d.god.canonical_tag(),
                                boon: d.boon.into(),
                                owed: d.owed.into(),
                                ticks_left: d.ticks_left,
                                altar_receipt: d.altar_receipt,
                            })
                            .collect(),
                    },
                )
            })
            .collect(),
        deep_weather: i.deep_weather,
        npcs: i
            .npcs
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    NpcWire {
                        name: v.name.into(),
                        job: v.job.canonical_tag(),
                        home: cell(v.home),
                        work: cell(v.work),
                        debt_cu: v.debt_cu,
                    },
                )
            })
            .collect(),
        shops: i
            .shops
            .into_iter()
            .map(|(k, v)| {
                (
                    cell(k),
                    ShopWire {
                        broke: v.broke.into_iter().collect(),
                        stole: v.stole.into_iter().collect(),
                        killed: v.killed.into_iter().collect(),
                    },
                )
            })
            .collect(),
        gossip: i
            .gossip
            .into_iter()
            .map(|v| GossipWire {
                from: v.from.into(),
                to: v.to.into(),
                sent_day: v.sent_day,
                ledger_delta: v.ledger_delta,
            })
            .collect(),
        chalk: i
            .chalk
            .marks
            .into_iter()
            .map(|(k, v)| (cell(k), v.into()))
            .collect(),
        mule: i.mule.map(|v| v.waypoints.into_iter().map(cell).collect()),
        vault_choice: i.vault_choice.map_or(0, VaultChoice::canonical_tag),
        chapel_debts: i
            .chapel_debts
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    ChapelDebtWire {
                        burials_owed: v.burials_owed,
                        chapel: v.chapel.into(),
                    },
                )
            })
            .collect(),
        next_id: i.next_id,
        step_mode: i.step_mode,
        pending: i
            .pending
            .iter()
            .map(pending_row_to_wire)
            .collect::<Result<_, _>>()?,
        observations: i
            .observations
            .iter()
            .map(observation_row_to_wire)
            .collect::<Result<_, _>>()?,
        branch_epoch: i.branch_epoch,
    })
}

fn wire_to_image(w: StateWire) -> Result<StateImage, PersistenceError> {
    limit(w.chunks.len(), MAX_CHUNKS, "state.chunks")?;
    tables_limit(&w)?;
    let mut strings = EncodeBudget::default();
    strings.check_state(&w)?;
    let chunks = w
        .chunks
        .into_iter()
        .enumerate()
        .map(|(i, c)| {
            if c.voxels.len() != CHUNK_VOL {
                return Err(PersistenceError::Invalid {
                    context: format!(
                        "state.chunks[{i}].voxels length {} != {CHUNK_VOL}",
                        c.voxels.len()
                    ),
                });
            }
            Ok(ChunkImage {
                coord: uncell(c.coord),
                voxels: c.voxels.into_iter().map(wire_to_voxel).collect(),
            })
        })
        .collect::<Result<_, _>>()?;
    Ok(StateImage {
        tick: w.tick,
        seed: w.seed,
        rng_master: w.rng_master,
        ecs_entity_count: w.ecs_entity_count,
        chunks,
        actors: w
            .actors
            .into_iter()
            .enumerate()
            .map(|(i, a)| {
                Ok(Actor {
                    id: a.id,
                    position: hexwork_core::spatial::FixedVec3::new(a.position_mm)
                        .map_err(|e| PersistenceError::Invalid { context: format!("state.actors[{i}].position_mm: {e}") })?,
                    body: hexwork_core::spatial::BodyExtents::new(a.body_mm[0], a.body_mm[1], a.body_mm[2])
                        .map_err(|e| PersistenceError::Invalid { context: format!("state.actors[{i}].body_mm: {e}") })?,
                    heading: a.heading.map(hexwork_core::heading::Heading::new).transpose().map_err(|_| PersistenceError::Invalid { context: format!("state.actors[{i}].heading") })?,
                    movement_remainder: hexwork_core::heading::MovementRemainder::new(a.movement_remainder).map_err(|_| PersistenceError::Invalid { context: format!("state.actors[{i}].movement_remainder") })?,
                    stance: stance(a.stance, &format!("state.actors[{i}].stance"))?,
                    hp: a.hp,
                    alive: a.alive,
                })
            })
            .collect::<Result<_, PersistenceError>>()?,
        actor_identities: identities_from(w.identities)?,
        wallets: rows_map(w.wallets, "state.wallets")?
            .into_iter()
            .map(|(k, v)| (k, Wallet::new(v.copper, v.scrip)))
            .collect(),
        contracts: contracts_from(w.contracts)?,
        run_location: location_from(w.location)?,
        connectors: w
            .connectors
            .into_iter()
            .enumerate()
            .map(|(i, v)| {
                Ok(Connector {
                    id: v.id,
                    kind: connector_kind(v.kind, &format!("state.connectors[{i}].kind"))?,
                    from: v.from.map(f32::from_bits),
                    to: v.to.map(f32::from_bits),
                    length: f32::from_bits(v.length),
                    anchor_quality: v.anchor_quality,
                    load_kg: f32::from_bits(v.load_kg),
                    blocked: v.blocked,
                })
            })
            .collect::<Result<_, PersistenceError>>()?,
        pressure_nodes: w
            .pressure_nodes
            .into_iter()
            .map(|v| PressureNode {
                id: v.id,
                pressure: f32::from_bits(v.pressure),
            })
            .collect(),
        pressure_edges: w
            .pressure_edges
            .into_iter()
            .map(|v| PressureEdge {
                from: v.from,
                to: v.to,
                openness: f32::from_bits(v.openness),
                hidden: v.hidden,
            })
            .collect(),
        causes: w
            .causes
            .into_iter()
            .map(|v| CauseImage {
                tick: v.tick,
                rule: v.rule.into_inner(),
                detail: v.detail.into_inner(),
            })
            .collect(),
        far: FarField {
            day: w.far.day,
            day_tick: w.far.day_tick,
        },
        laws: w
            .laws
            .into_iter()
            .map(|v| LawImage {
                rule: v.rule.into_inner(),
                first_seen_tick: v.tick,
            })
            .collect(),
        fall_odo: rows_map(w.fall_odo, "state.fall_odo")?,
        lights: w
            .lights
            .into_iter()
            .map(|v| LightSource {
                x: v.x,
                y: v.y,
                z: v.z,
                radius: v.radius,
                warm: v.warm,
            })
            .collect(),
        sounds: w
            .sounds
            .into_iter()
            .map(|v| SoundEvent {
                x: v.x,
                y: v.y,
                z: v.z,
                loudness: v.loudness,
                timbre: v.timbre,
            })
            .collect(),
        corpses: rows_map(w.corpses, "state.corpses")?
            .into_iter()
            .map(|(k, v)| {
                (
                    uncell(k),
                    CorpseClock {
                        age_ticks: v.age_ticks,
                        bursts_at: v.bursts_at,
                    },
                )
            })
            .collect(),
        signal_sources: w.signal_sources.into_iter().map(uncell).collect(),
        spill: Spill {
            desired_moves: {
                let mut rows = Vec::new();
                for (id,heading,speed_mm_s,stop_at) in w.spill.desired_moves {
                    let heading = heading.map(hexwork_core::heading::Heading::new).transpose().map_err(|_| PersistenceError::Invalid { context: "state.spill.desired_moves.heading".into() })?;
                    let stop_at=stop_at.map(hexwork_core::spatial::FixedVec3::new).transpose().map_err(|_|PersistenceError::Invalid {context:"state.spill.desired_moves.stop_at".into()})?;
                    let sample = hexwork_core::locomotion::MovementSample { heading,speed_mm_s,stop_at };
                    if !sample.valid() { return Err(PersistenceError::Invalid { context: "state.spill.desired_moves.speed".into() }); }
                    rows.push((id,sample));
                }
                rows_map(rows, "state.spill.desired_moves")?
            },
            crossed_cells: {
                let mut result = BTreeMap::new();
                for (id,cells) in rows_map(w.spill.crossed_cells, "state.spill.crossed_cells")? {
                    let mut set = std::collections::BTreeSet::new();
                    for c in cells { if !set.insert(uncell(c)) {return Err(PersistenceError::Invalid {context:"duplicate crossed cell".into()});} }
                    result.insert(id,set);
                }
                result
            },
            gas_flows: w
                .spill
                .gas_flows
                .into_iter()
                .map(|(a, b, c)| (a, b, f32::from_bits(c)))
                .collect(),
            light_pending: w.spill.light_pending,
            signal_pending: w.spill.signal_pending,
            scent_pending: w.spill.scent_pending,
        },
        characters: characters_from(w.characters)?,
        reputations: reputations_from(w.reputations)?,
        items: items_from(w.items, "state.items")?,
        ground_items: rows_map(w.ground_items, "state.ground_items")?
            .into_iter()
            .map(|(k, v)| (k, uncell(v)))
            .collect(),
        packs: packs_from(w.packs)?,
        knowledge: knowledge_from_wire(w.knowledge)?,
        idf: rows_map(w.idf, "state.idf")?
            .into_iter()
            .map(|(k, v)| (k.into_inner(), v))
            .collect(),
        hymn_cases: hymns_from(w.hymn_cases)?,
        cast_queue: w
            .cast_queue
            .into_iter()
            .map(|v| {
                (
                    v.caster,
                    v.template.into_inner(),
                    [v.target.x, v.target.y, v.target.z],
                )
            })
            .collect(),
        bargains: bargains_from(w.bargains)?,
        deep_weather: w.deep_weather,
        npcs: npcs_from(w.npcs)?,
        shops: shops_from(w.shops)?,
        gossip: w
            .gossip
            .into_iter()
            .map(|v| GossipPacket {
                from: v.from.into_inner(),
                to: v.to.into_inner(),
                sent_day: v.sent_day,
                ledger_delta: v.ledger_delta,
            })
            .collect(),
        chalk: Chalk {
            marks: rows_map(w.chalk, "state.chalk")?
                .into_iter()
                .map(|(k, v)| (uncell(k), v.into_inner()))
                .collect(),
        },
        mule: w.mule.map(|v| MulePath {
            waypoints: v.into_iter().map(uncell).collect(),
        }),
        vault_choice: vault_choice(w.vault_choice)?,
        chapel_debts: rows_map(w.chapel_debts, "state.chapel_debts")?
            .into_iter()
            .map(|(k, v)| {
                (
                    k,
                    ChapelDebt {
                        burials_owed: v.burials_owed,
                        chapel: v.chapel.into_inner(),
                    },
                )
            })
            .collect(),
        next_id: w.next_id,
        step_mode: w.step_mode,
        pending: w
            .pending
            .into_inner()
            .into_iter()
            .enumerate()
            .map(|(i, row)| pending_row_from_wire(row, &format!("state.pending[{i}]")))
            .collect::<Result<_, _>>()?,
        observations: {
            let rows = w
                .observations
                .into_inner()
                .into_iter()
                .enumerate()
                .map(|(i, row)| {
                    observation_row_from_wire(row, &format!("state.observations[{i}]"))
                })
                .collect::<Result<Vec<_>, _>>()?;
            require_strict_observation_order(&rows)?;
            rows
        },
        branch_epoch: w.branch_epoch,
    })
}

fn tables_limit(w: &StateWire) -> Result<(), PersistenceError> {
    for (n, l) in [
        ("actors", w.actors.len()),
        ("identities", w.identities.len()),
        ("wallets", w.wallets.len()),
        ("contracts", w.contracts.len()),
        ("connectors", w.connectors.len()),
        ("pressure_nodes", w.pressure_nodes.len()),
        ("pressure_edges", w.pressure_edges.len()),
        ("causes", w.causes.len()),
        ("laws", w.laws.len()),
        ("fall_odo", w.fall_odo.len()),
        ("spill.desired_moves", w.spill.desired_moves.len()),
        ("spill.crossed_cells", w.spill.crossed_cells.len()),
        ("lights", w.lights.len()),
        ("sounds", w.sounds.len()),
        ("corpses", w.corpses.len()),
        ("signal_sources", w.signal_sources.len()),
        ("characters", w.characters.len()),
        ("reputations", w.reputations.len()),
        ("items", w.items.len()),
        ("ground_items", w.ground_items.len()),
        ("packs", w.packs.len()),
        ("idf", w.idf.len()),
        ("hymn_cases", w.hymn_cases.len()),
        ("cast_queue", w.cast_queue.len()),
        ("bargains", w.bargains.len()),
        ("npcs", w.npcs.len()),
        ("shops", w.shops.len()),
        ("gossip", w.gossip.len()),
        ("chalk", w.chalk.len()),
        ("chapel_debts", w.chapel_debts.len()),
        ("pending", w.pending.len()),
        ("observations", w.observations.len()),
    ] {
        limit(l, MAX_ROWS, &format!("state.{n}"))?;
    }
    Ok(())
}

#[derive(Default)]
struct EncodeBudget {
    strings: usize,
    chunks: usize,
    state_rows: usize,
    nested_rows: usize,
    gas_flows: usize,
    mule_waypoints: usize,
    town_ledger: usize,
    pack_items: usize,
    hymn_cards: usize,
    hymn_miscasts: usize,
    ledger_debts: usize,
    shop_rows: usize,
    item_tags: usize,
    initial_rows: usize,
    initial_tags: usize,
    initial_strings: usize,
    knowledge_rows: usize,
    knowledge_strings: usize,
}
impl EncodeBudget {
    fn check_identity(
        &mut self,
        identity: &RunIdentity,
        context: &str,
    ) -> Result<(), PersistenceError> {
        self.add(
            &identity.display_name,
            &format!("{context}.display_name"),
            true,
        )?;
        self.add(
            &identity.content_hash,
            &format!("{context}.content_hash"),
            true,
        )?;
        validate_initial_lineage(&identity.initial_lineage, true).map_err(|e| {
            PersistenceError::Invalid {
                context: format!("{context}.initial_lineage: {e}"),
            }
        })?;
        let initial = &identity.initial_lineage;
        let initial_rows = initial.mapped_regions.len()
            + initial.recipes.len()
            + initial.translated_hymns.len()
            + initial.identified_item_templates.len()
            + initial.ghosts.len()
            + initial.auction_lots.len();
        self.add_initial_rows(initial_rows)?;
        self.state_rows = checked_value(
            self.state_rows,
            initial_rows,
            MAX_TOTAL_STATE_ROWS,
            "total state rows",
        )?;
        self.nested_rows = checked_value(
            self.nested_rows,
            initial_rows,
            MAX_TOTAL_NESTED_ROWS,
            "total nested rows",
        )?;
        for id in initial
            .mapped_regions
            .iter()
            .chain(&initial.recipes)
            .chain(&initial.translated_hymns)
            .chain(&initial.identified_item_templates)
        {
            self.add_initial(id, "initial_lineage.id")?;
        }
        for ghost in &initial.ghosts {
            self.add_initial(&ghost.item_template_id, "initial_lineage.ghost.template")?;
            self.add_initial_text(
                &ghost.warning,
                "initial_lineage.ghost.warning",
                MAX_GHOST_WARNING_BYTES,
            )?;
        }
        for lot in &initial.auction_lots {
            self.add_initial(&lot.template_id, "initial_lineage.lot.template")?;
            self.initial_tags = checked_value(
                self.initial_tags,
                lot.tags.len(),
                MAX_INITIAL_TAGS,
                "initial lineage tags",
            )?;
            self.nested_rows = checked_value(
                self.nested_rows,
                lot.tags.len(),
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
            for tag in &lot.tags {
                self.add_initial_text(tag, "initial_lineage.lot.tag", MAX_LOT_TAG_BYTES)?;
            }
        }
        Ok(())
    }

    fn check_events(
        &mut self,
        rows: &[SessionEvent],
        context: &str,
    ) -> Result<(), PersistenceError> {
        self.check_event_view(EventView::new(rows, None)?, context)
    }

    fn check_event_view(
        &mut self,
        rows: EventView<'_>,
        context: &str,
    ) -> Result<(), PersistenceError> {
        limit(rows.checked_len()?, MAX_SESSION_EVENTS, context)?;
        for (index, row) in rows.iter().enumerate() {
            if let SessionEventKind::Action(action) = &row.kind {
                let text = match action {
                    CommittedAction::InspectBoard { town } => Some(town.as_str()),
                    CommittedAction::AcceptContract { id } => Some(id.as_str()),
                    CommittedAction::TurnIn { contract_id } => Some(contract_id.as_str()),
                    CommittedAction::Move(_)
                    | CommittedAction::Wait
                    | CommittedAction::UseTransition { .. }
                    | CommittedAction::ResolveAttack { .. }
                    | CommittedAction::Pickup { .. }
                    | CommittedAction::Drop { .. }
                    | CommittedAction::Wield { .. }
                    | CommittedAction::TinkerIdentify { .. }
                    | CommittedAction::UseService { .. } => None,
                };
                if let Some(text) = text {
                    self.add(text, &format!("{context}[{index}].action.text"), true)?;
                }
            }
        }
        Ok(())
    }
    fn add_initial_rows(&mut self, rows: usize) -> Result<(), PersistenceError> {
        self.initial_rows = checked_value(
            self.initial_rows,
            rows,
            MAX_INITIAL_ROWS,
            "initial lineage rows",
        )?;
        Ok(())
    }
    fn add_initial(&mut self, s: &str, c: &str) -> Result<(), PersistenceError> {
        self.add_initial_text(s, c, KNOWLEDGE_ID_MAX_BYTES)
    }
    fn add_knowledge(&mut self, s: &str, c: &str) -> Result<(), PersistenceError> {
        limit(s.len(), KNOWLEDGE_ID_MAX_BYTES, c)?;
        self.add(s, c, true)?;
        self.knowledge_strings = checked_value(
            self.knowledge_strings,
            s.len(),
            KNOWLEDGE_TOTAL_ID_BYTES_LIMIT,
            "working knowledge strings",
        )?;
        Ok(())
    }
    fn add_initial_text(&mut self, s: &str, c: &str, cap: usize) -> Result<(), PersistenceError> {
        limit(s.len(), cap, c)?;
        self.add(s, c, true)?;
        self.initial_strings = checked_value(
            self.initial_strings,
            s.len(),
            MAX_INITIAL_STRING_BYTES,
            "initial lineage strings",
        )?;
        Ok(())
    }
    fn add(&mut self, s: &str, c: &str, blank: bool) -> Result<(), PersistenceError> {
        valid_string(s, c, blank)?;
        self.strings =
            self.strings
                .checked_add(s.len())
                .ok_or_else(|| PersistenceError::Limit {
                    context: "total strings".into(),
                    actual: usize::MAX,
                    limit: MAX_TOTAL_STRING_BYTES,
                })?;
        limit(self.strings, MAX_TOTAL_STRING_BYTES, "total strings")
    }
    fn check_image(&mut self, i: &StateImage) -> Result<(), PersistenceError> {
        validate_image_limits(i)?;
        checked_total(
            &mut self.chunks,
            i.chunks.len(),
            MAX_TOTAL_CHUNKS,
            "total chunks",
        )?;
        for rows in [
            i.chunks.len(),
            i.actors.len(),
            i.actor_identities.len(),
            i.wallets.len(),
            i.contracts.len(),
            i.connectors.len(),
            i.pressure_nodes.len(),
            i.pressure_edges.len(),
            i.causes.len(),
            i.laws.len(),
            i.fall_odo.len(),
            i.spill.desired_moves.len(),
            i.spill.crossed_cells.len(),
            i.lights.len(),
            i.sounds.len(),
            i.corpses.len(),
            i.signal_sources.len(),
            i.characters.len(),
            i.reputations.len(),
            i.items.len(),
            i.ground_items.len(),
            i.packs.len(),
            i.knowledge.mapped_regions.len(),
            i.knowledge.recipes.len(),
            i.knowledge.translated_hymns.len(),
            i.idf.len(),
            i.hymn_cases.len(),
            i.cast_queue.len(),
            i.bargains.len(),
            i.npcs.len(),
            i.shops.len(),
            i.gossip.len(),
            i.chalk.marks.len(),
            i.chapel_debts.len(),
            i.pending.len(),
        ] {
            checked_total(
                &mut self.state_rows,
                rows,
                MAX_TOTAL_STATE_ROWS,
                "total state rows",
            )?;
        }
        for cells in i.spill.crossed_cells.values() {
            limit(cells.len(), MAX_ROWS, "spill.crossed_cells.cells")?;
            checked_total(&mut self.nested_rows,cells.len(),MAX_TOTAL_NESTED_ROWS,"total nested rows")?;
        }
        for chunk in &i.chunks {
            checked_total(
                &mut self.nested_rows,
                chunk.voxels.len(),
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
        }
        let gas_flows = i.spill.gas_flows.len();
        checked_total(
            &mut self.gas_flows,
            gas_flows,
            MAX_TOTAL_GAS_FLOWS,
            "spill gas flows",
        )?;
        checked_total(
            &mut self.nested_rows,
            gas_flows,
            MAX_TOTAL_NESTED_ROWS,
            "total nested rows",
        )?;
        let mule = i.mule.as_ref().map_or(0, |m| m.waypoints.len());
        checked_total(
            &mut self.mule_waypoints,
            mule,
            MAX_TOTAL_MULE_WAYPOINTS,
            "mule waypoints",
        )?;
        checked_total(
            &mut self.nested_rows,
            mule,
            MAX_TOTAL_NESTED_ROWS,
            "total nested rows",
        )?;
        for reputation in i.reputations.values() {
            let rows = reputation.town_ledger.len();
            checked_total(
                &mut self.town_ledger,
                rows,
                MAX_TOTAL_TOWN_LEDGER_ROWS,
                "reputation town-ledger rows",
            )?;
            checked_total(
                &mut self.nested_rows,
                rows,
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
        }
        for item in i.items.values() {
            self.add_item_rows(item)?;
        }
        for pack in i.packs.values() {
            checked_total(
                &mut self.pack_items,
                pack.items.len(),
                MAX_TOTAL_PACK_ITEMS,
                "pack items",
            )?;
            checked_total(
                &mut self.nested_rows,
                pack.items.len(),
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
            for item in pack.items.values() {
                self.add_item_rows(item)?;
            }
        }
        for case in i.hymn_cases.values() {
            checked_total(
                &mut self.hymn_cards,
                case.cards.len(),
                MAX_TOTAL_HYMN_CARDS,
                "hymn cards",
            )?;
            checked_total(
                &mut self.nested_rows,
                case.cards.len(),
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
            for card in &case.cards {
                checked_total(
                    &mut self.hymn_miscasts,
                    card.miscasts.len(),
                    MAX_TOTAL_HYMN_MISCASTS,
                    "hymn miscasts",
                )?;
                checked_total(
                    &mut self.nested_rows,
                    card.miscasts.len(),
                    MAX_TOTAL_NESTED_ROWS,
                    "total nested rows",
                )?;
            }
        }
        for ledger in i.bargains.values() {
            checked_total(
                &mut self.ledger_debts,
                ledger.debts.len(),
                MAX_TOTAL_LEDGER_DEBTS,
                "ledger debts",
            )?;
            checked_total(
                &mut self.nested_rows,
                ledger.debts.len(),
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
        }
        for shop in i.shops.values() {
            let rows = shop
                .broke
                .len()
                .checked_add(shop.stole.len())
                .and_then(|v| v.checked_add(shop.killed.len()))
                .ok_or_else(|| PersistenceError::Invalid {
                    context: "shop row count overflows".into(),
                })?;
            checked_total(
                &mut self.shop_rows,
                rows,
                MAX_TOTAL_SHOP_ROWS,
                "shop memory rows",
            )?;
            checked_total(
                &mut self.nested_rows,
                rows,
                MAX_TOTAL_NESTED_ROWS,
                "total nested rows",
            )?;
        }
        self.check_image_strings(i)
    }

    fn check_image_strings(&mut self, i: &StateImage) -> Result<(), PersistenceError> {
        for value in i.actor_identities.values() {
            self.add(&value.template_id, "identity.template", true)?;
            self.add(&value.display_name, "identity.name", true)?;
        }
        for (key, value) in &i.contracts {
            self.add(key, "contract.key", true)?;
            self.add(&value.id, "contract.id", true)?;
            self.add(&value.reward_town, "contract.reward_town", true)?;
        }
        if let RunLocation::Town(town) = &i.run_location {
            self.add(town, "location.town", true)?;
        }
        for value in &i.causes {
            self.add(&value.rule, "cause.rule", true)?;
            self.add(&value.detail, "cause.detail", false)?;
        }
        for value in &i.laws {
            self.add(&value.rule, "law.rule", true)?;
        }
        for value in i.reputations.values() {
            for town in value.town_ledger.keys() {
                self.add(town, "reputation.town", true)?;
            }
        }
        for value in i.items.values() {
            self.image_item(value)?;
        }
        for pack in i.packs.values() {
            for value in pack.items.values() {
                self.image_item(value)?;
            }
        }
        let knowledge_rows = i.knowledge.mapped_regions.len()
            + i.knowledge.recipes.len()
            + i.knowledge.translated_hymns.len();
        self.knowledge_rows = checked_value(
            self.knowledge_rows,
            knowledge_rows,
            KNOWLEDGE_TOTAL_ROW_LIMIT,
            "working knowledge rows",
        )?;
        for (name, rows) in [
            ("knowledge.mapped_regions", &i.knowledge.mapped_regions),
            ("knowledge.recipes", &i.knowledge.recipes),
            ("knowledge.translated_hymns", &i.knowledge.translated_hymns),
        ] {
            for id in rows {
                self.add_knowledge(id, name)?;
            }
        }
        for template in i.idf.keys() {
            self.add(template, "idf.template", true)?;
        }
        for case in i.hymn_cases.values() {
            for card in &case.cards {
                self.add(&card.template, "hymn.template", true)?;
                for miscast in &card.miscasts {
                    self.add(miscast, "hymn.miscast", true)?;
                }
            }
        }
        for (_, template, _) in &i.cast_queue {
            self.add(template, "cast.template", true)?;
        }
        for ledger in i.bargains.values() {
            for debt in &ledger.debts {
                self.add(&debt.boon, "debt.boon", true)?;
                self.add(&debt.owed, "debt.owed", true)?;
            }
        }
        for npc in i.npcs.values() {
            self.add(&npc.name, "npc.name", true)?;
        }
        for gossip in &i.gossip {
            self.add(&gossip.from, "gossip.from", true)?;
            self.add(&gossip.to, "gossip.to", true)?;
        }
        for text in i.chalk.marks.values() {
            self.add(text, "chalk.text", true)?;
        }
        for debt in i.chapel_debts.values() {
            self.add(&debt.chapel, "chapel_debt.chapel", true)?;
        }
        for pending in &i.pending {
            if let Some(text) = pending_text(pending) {
                self.add(text, "pending.text", true)?;
            }
        }
        Ok(())
    }

    fn add_item_rows(&mut self, value: &Instance) -> Result<(), PersistenceError> {
        checked_total(
            &mut self.item_tags,
            value.tags.len(),
            MAX_TOTAL_ITEM_TAGS,
            "item tags",
        )?;
        checked_total(
            &mut self.nested_rows,
            value.tags.len(),
            MAX_TOTAL_NESTED_ROWS,
            "total nested rows",
        )
    }

    fn image_item(&mut self, value: &Instance) -> Result<(), PersistenceError> {
        self.add(&value.template, "item.template", true)?;
        limit(value.tags.len(), MAX_ITEM_TAGS, "item.tags")?;
        for tag in &value.tags {
            self.add(tag, "item.tag", true)?;
        }
        Ok(())
    }

    fn check_state(&mut self, w: &StateWire) -> Result<(), PersistenceError> {
        for (_, v) in &w.identities {
            self.add(&v.template, "identity.template", true)?;
            self.add(&v.name, "identity.name", true)?;
        }
        for (k, v) in &w.contracts {
            self.add(k, "contract.key", true)?;
            self.add(&v.id, "contract.id", true)?;
            self.add(&v.reward_town, "contract.reward_town", true)?;
        }
        if w.location.tag == 1 {
            self.add(&w.location.town, "location.town", true)?;
        }
        for v in &w.causes {
            self.add(&v.rule, "cause.rule", true)?;
            self.add(&v.detail, "cause.detail", false)?;
        }
        for v in &w.laws {
            self.add(&v.rule, "law.rule", true)?;
        }
        for (_, v) in &w.reputations {
            for (k, _) in &v.town_ledger {
                self.add(k, "reputation.town", true)?;
            }
        }
        for (_, v) in &w.items {
            self.item(v)?;
        }
        for (_, p) in &w.packs {
            for (_, v) in &p.items {
                self.item(v)?;
            }
        }
        for id in w
            .knowledge
            .mapped_regions
            .iter()
            .chain(&w.knowledge.recipes)
            .chain(&w.knowledge.translated_hymns)
        {
            self.add_knowledge(&id.0, "knowledge.id")?;
        }
        for (k, _) in &w.idf {
            self.add(k, "idf.template", true)?;
        }
        for (_, h) in &w.hymn_cases {
            for c in &h.cards {
                self.add(&c.template, "hymn.template", true)?;
                for m in &c.miscasts {
                    self.add(m, "hymn.miscast", true)?;
                }
            }
        }
        for c in &w.cast_queue {
            self.add(&c.template, "cast.template", true)?;
        }
        for (_, l) in &w.bargains {
            for d in &l.debts {
                self.add(&d.boon, "debt.boon", true)?;
                self.add(&d.owed, "debt.owed", true)?;
            }
        }
        for (_, n) in &w.npcs {
            self.add(&n.name, "npc.name", true)?;
        }
        for g in &w.gossip {
            self.add(&g.from, "gossip.from", true)?;
            self.add(&g.to, "gossip.to", true)?;
        }
        for (_, s) in &w.chalk {
            self.add(s, "chalk.text", true)?;
        }
        for (_, d) in &w.chapel_debts {
            self.add(&d.chapel, "chapel_debt.chapel", true)?;
        }
        for row in &w.pending {
            if !row.text.is_empty() {
                self.add(&row.text, "pending.text", true)?;
            }
        }
        for row in &w.observations {
            if !row.town.is_empty() {
                self.add(&row.town, "observation.town", true)?;
            }
        }
        Ok(())
    }
    fn item(&mut self, v: &ItemWire) -> Result<(), PersistenceError> {
        self.add(&v.template, "item.template", true)?;
        limit(v.tags.len(), 256, "item.tags")?;
        for t in &v.tags {
            self.add(t, "item.tag", true)?;
        }
        Ok(())
    }
}

fn rows_map<K, V, I>(rows: I, table: &str) -> Result<BTreeMap<K, V>, PersistenceError>
where
    K: Ord + Clone,
    I: IntoIterator<Item = (K, V)>,
    I::IntoIter: ExactSizeIterator,
{
    let rows = rows.into_iter();
    limit(rows.len(), MAX_ROWS, table)?;
    let mut out = BTreeMap::new();
    let mut prior = None;
    for (i, (k, v)) in rows.enumerate() {
        if prior.as_ref().is_some_and(|p| k <= *p) {
            return Err(PersistenceError::Invalid {
                context: format!("{table}[{i}] is duplicate or noncanonical order"),
            });
        }
        prior = Some(k.clone());
        out.insert(k, v);
    }
    Ok(out)
}
fn identities_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, IdentityWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, ActorIdentity>, PersistenceError> {
    rows_map(rows, "state.identities")?
        .into_iter()
        .enumerate()
        .map(|(i, (key, value))| {
            Ok((
                key,
                ActorIdentity::new(
                    value.template,
                    value.name,
                    role(value.role, &format!("state.identities[{i}].role"))?,
                ),
            ))
        })
        .collect()
}

fn contracts_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(BoundedString, ContractWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<String, ActiveContract>, PersistenceError> {
    rows_map(rows, "state.contracts")?
        .into_iter()
        .map(|(k, v)| {
            Ok((
                k.into_inner(),
                ActiveContract {
                    id: v.id.into_inner(),
                    taker: v.taker,
                    accepted_tick: v.accepted_tick,
                    floor: v.floor,
                    pay_cu: v.pay_cu,
                    objective_target: v.target,
                    objective_progress: v.progress,
                    status: contract_status(v.status, "contract.status")?,
                    reward_town: v.reward_town.into_inner(),
                    standing_delta: v.standing_delta,
                },
            ))
        })
        .collect::<Result<_, PersistenceError>>()
}
fn characters_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, CharacterWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, Character>, PersistenceError> {
    rows_map(rows, "state.characters")?
        .into_iter()
        .map(|(k, v)| {
            Ok((
                k,
                Character {
                    ancestry: ancestry(v.ancestry, "character.ancestry")?,
                    calling: calling(v.calling, "character.calling")?,
                    attrs: v.attrs,
                    skills: v.skills,
                    uses: v.uses,
                    pips: v.pips,
                },
            ))
        })
        .collect()
}
fn reputations_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, ReputationWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, Reputation>, PersistenceError> {
    rows_map(rows, "state.reputations")?
        .into_iter()
        .map(|(k, v)| {
            let fs = v
                .factions
                .into_iter()
                .enumerate()
                .map(|(i, s)| {
                    let expected = i as u8;
                    if s.faction != expected {
                        return Err(PersistenceError::Invalid {
                            context: format!(
                                "state.reputations[{k}].factions[{i}] expected tag {expected}, found {}",
                                s.faction
                            ),
                        });
                    }
                    Ok((
                        faction(s.faction, &format!("reputation.factions[{i}].tag"))?,
                        Standing {
                            rank: s.rank,
                            debt_cu: s.debt_cu,
                        },
                    ))
                })
                .collect::<Result<Vec<_>, PersistenceError>>()?
                .try_into()
                .map_err(|_| PersistenceError::Invalid {
                    context: "reputation factions length".into(),
                })?;
            let gs = v
                .god_credit
                .into_iter()
                .enumerate()
                .map(|(i, g)| {
                    let expected = i as u8;
                    if g.god != expected {
                        return Err(PersistenceError::Invalid {
                            context: format!(
                                "state.reputations[{k}].god_credit[{i}] expected tag {expected}, found {}",
                                g.god
                            ),
                        });
                    }
                    Ok((
                        god(g.god, &format!("reputation.gods[{i}].tag"))?,
                        g.credit,
                    ))
                })
                .collect::<Result<Vec<_>, PersistenceError>>()?
                .try_into()
                .map_err(|_| PersistenceError::Invalid {
                    context: "reputation gods length".into(),
                })?;
            Ok((
                k,
                Reputation {
                    factions: fs,
                    town_ledger: rows_map(v.town_ledger, "reputation.town_ledger")?
                        .into_iter()
                        .map(|(town, standing)| (town.into_inner(), standing))
                        .collect(),
                    god_credit: gs,
                },
            ))
        })
        .collect()
}
fn items_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, ItemWire), LIMIT, BUDGET>,
    table: &str,
) -> Result<BTreeMap<u64, Instance>, PersistenceError> {
    rows_map(rows, table)?
        .into_iter()
        .map(|(k, v)| Ok((k, item_from_wire(v)?)))
        .collect()
}
fn packs_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, PackWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, Pack>, PersistenceError> {
    rows_map(rows, "state.packs")?
        .into_iter()
        .map(|(k, v)| {
            Ok((
                k,
                Pack {
                    items: items_from(v.items, "pack.items")?,
                    equipped_weapon: v.equipped_weapon,
                    equipped_armor: v.equipped_armor,
                    max_weight_kg: f32::from_bits(v.max_weight_kg),
                    max_bulk: v.max_bulk,
                },
            ))
        })
        .collect()
}
fn hymns_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, HymnCaseWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, HymnCase>, PersistenceError> {
    rows_map(rows, "state.hymn_cases")?
        .into_iter()
        .map(|(k, v)| {
            limit(v.cards.len(), 256, "hymn_case.cards")?;
            Ok((
                k,
                HymnCase {
                    cards: v
                        .cards
                        .into_iter()
                        .map(|c| HymnCard {
                            template: c.template.into_inner(),
                            conduct_base: c.conduct_base,
                            slots: c.slots,
                            fragile: c.fragile,
                            miscasts: c
                                .miscasts
                                .into_iter()
                                .map(BoundedString::into_inner)
                                .collect(),
                        })
                        .collect(),
                    max_slots: v.max_slots,
                },
            ))
        })
        .collect()
}
fn bargains_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, LedgerWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, Ledger>, PersistenceError> {
    rows_map(rows, "state.bargains")?
        .into_iter()
        .map(|(k, v)| {
            limit(v.debts.len(), MAX_ROWS, "ledger.debts")?;
            Ok((
                k,
                Ledger {
                    debts: v
                        .debts
                        .into_iter()
                        .map(|d| {
                            Ok(Debt {
                                god: god(d.god, "debt.god")?,
                                boon: d.boon.into_inner(),
                                owed: d.owed.into_inner(),
                                ticks_left: d.ticks_left,
                                altar_receipt: d.altar_receipt,
                            })
                        })
                        .collect::<Result<_, PersistenceError>>()?,
                },
            ))
        })
        .collect()
}
fn npcs_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(u64, NpcWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<u64, Townsfolk>, PersistenceError> {
    rows_map(rows, "state.npcs")?
        .into_iter()
        .map(|(k, v)| {
            Ok((
                k,
                Townsfolk {
                    name: v.name.into_inner(),
                    job: job(v.job, "npc.job")?,
                    home: uncell(v.home),
                    work: uncell(v.work),
                    debt_cu: v.debt_cu,
                },
            ))
        })
        .collect()
}
fn shops_from<const LIMIT: usize, const BUDGET: u8>(
    rows: BoundedVec<(CellWire, ShopWire), LIMIT, BUDGET>,
) -> Result<BTreeMap<(i32, i32, i32), ShopMemory>, PersistenceError> {
    rows_map(rows, "state.shops")?
        .into_iter()
        .map(|(k, v)| {
            Ok((
                uncell(k),
                ShopMemory {
                    broke: rows_map(v.broke, "shop.broke")?,
                    stole: rows_map(v.stole, "shop.stole")?,
                    killed: rows_map(v.killed, "shop.killed")?,
                },
            ))
        })
        .collect()
}

fn cell(v: (i32, i32, i32)) -> CellWire {
    CellWire {
        x: v.0,
        y: v.1,
        z: v.2,
    }
}
fn uncell(v: CellWire) -> (i32, i32, i32) {
    (v.x, v.y, v.z)
}
fn voxel_to_wire(v: Voxel) -> VoxelWire {
    VoxelWire {
        material: v.material,
        occ: v.occ,
        fluid_kind: v.fluid_kind,
        fluid_amt: v.fluid_amt,
        gas_kind: v.gas_kind,
        gas_amt: v.gas_amt,
        light: v.light,
        scent: v.scent,
        flags: v.flags,
        pad: v.pad,
        pressure: v.pressure,
        flow: v.flow,
        extra: v.extra,
    }
}
fn wire_to_voxel(v: VoxelWire) -> Voxel {
    Voxel {
        material: v.material,
        occ: v.occ,
        fluid_kind: v.fluid_kind,
        fluid_amt: v.fluid_amt,
        gas_kind: v.gas_kind,
        gas_amt: v.gas_amt,
        light: v.light,
        scent: v.scent,
        flags: v.flags,
        pad: v.pad,
        pressure: v.pressure,
        flow: v.flow,
        extra: v.extra,
    }
}
fn item_to_wire(v: Instance) -> ItemWire {
    ItemWire {
        id: v.id,
        template: v.template.into(),
        material: v.material,
        identified: v.identified,
        cursed: v.cursed,
        blessed: v.blessed,
        condition: v.condition.canonical_tag(),
        silvered: v.silvered,
        illegal: v.illegal,
        tags: v.tags.into_iter().map(Into::into).collect(),
        fuel: v.fuel,
        wet_ticks: v.wet_ticks,
    }
}
fn item_from_wire(v: ItemWire) -> Result<Instance, PersistenceError> {
    Ok(Instance {
        id: v.id,
        template: v.template.into_inner(),
        material: v.material,
        identified: v.identified,
        cursed: v.cursed,
        blessed: v.blessed,
        condition: condition(v.condition, "item.condition")?,
        silvered: v.silvered,
        illegal: v.illegal,
        tags: v.tags.into_iter().map(BoundedString::into_inner).collect(),
        fuel: v.fuel,
        wet_ticks: v.wet_ticks,
    })
}
fn pack_to_wire(v: Pack) -> PackWire {
    PackWire {
        items: v
            .items
            .into_iter()
            .map(|(k, v)| (k, item_to_wire(v)))
            .collect(),
        equipped_weapon: v.equipped_weapon,
        equipped_armor: v.equipped_armor,
        max_weight_kg: v.max_weight_kg.to_bits(),
        max_bulk: v.max_bulk,
    }
}
fn reputation_to_wire(v: Reputation) -> ReputationWire {
    ReputationWire {
        factions: v.factions.map(|(f, s)| StandingWire {
            faction: f.canonical_tag(),
            rank: s.rank,
            debt_cu: s.debt_cu,
        }),
        town_ledger: v
            .town_ledger
            .into_iter()
            .map(|(town, standing)| (town.into(), standing))
            .collect(),
        god_credit: v.god_credit.map(|(g, c)| GodCreditWire {
            god: g.canonical_tag(),
            credit: c,
        }),
    }
}
fn location_to_wire(v: RunLocation) -> LocationWire {
    match v {
        RunLocation::Uninitialized => LocationWire {
            tag: 0,
            town: BoundedString::default(),
            depth: 0,
        },
        RunLocation::Town(town) => LocationWire {
            tag: 1,
            town: town.into(),
            depth: 0,
        },
        RunLocation::Deep(depth) => LocationWire {
            tag: 2,
            town: BoundedString::default(),
            depth,
        },
    }
}
fn location_from(v: LocationWire) -> Result<RunLocation, PersistenceError> {
    match v.tag {
        0 if v.town.is_empty() && v.depth == 0 => Ok(RunLocation::Uninitialized),
        1 if v.depth == 0 && !v.town.trim().is_empty() => {
            Ok(RunLocation::Town(v.town.into_inner()))
        }
        2 if v.town.is_empty() && v.depth > 0 => Ok(RunLocation::Deep(v.depth)),
        _ => Err(PersistenceError::Invalid {
            context: "state.location tag/payload is noncanonical".into(),
        }),
    }
}

macro_rules! tagged {
    ($name:ident, $ty:ty, $ctx:ident, {$($tag:literal => $value:path),+ $(,)?}) => {
        fn $name(tag: u8, $ctx: &str) -> Result<$ty, PersistenceError> {
            match tag {
                $($tag => Ok($value),)+
                _ => Err(PersistenceError::Invalid {
                    context: format!("{} unknown tag {tag}", $ctx),
                }),
            }
        }
    };
}
tagged!(stance,Stance,c,{0=>Stance::Stand,1=>Stance::Crouch,2=>Stance::Climb,3=>Stance::Swim,4=>Stance::Prone,5=>Stance::Rope});
tagged!(role,ActorRole,c,{0=>ActorRole::Player,1=>ActorRole::Npc,2=>ActorRole::Monster});
tagged!(contract_status,ContractStatus,c,{0=>ContractStatus::Active,1=>ContractStatus::Completed,2=>ContractStatus::TurnedIn,3=>ContractStatus::Failed});
tagged!(connector_kind,ConnectorKind,c,{0=>ConnectorKind::Stairwell,1=>ConnectorKind::Shaft,2=>ConnectorKind::Well,3=>ConnectorKind::Vent,4=>ConnectorKind::ElevatorCar,5=>ConnectorKind::Counterweight,6=>ConnectorKind::RopeAnchor});
tagged!(ancestry,Ancestry,c,{0=>Ancestry::Mudborn,1=>Ancestry::CantBorn,2=>Ancestry::PaleTouched,3=>Ancestry::Foundling});
tagged!(calling,Calling,c,{0=>Calling::Ratcatcher,1=>Calling::Lampbearer,2=>Calling::Tinker,3=>Calling::CorpseTender,4=>Calling::TollWard,5=>Calling::ChoirDeserter});
tagged!(faction,Faction,c,{0=>Faction::Lampwrights,1=>Faction::Congregation,2=>Faction::Wardens,3=>Faction::Audit,4=>Faction::Choir});
tagged!(god,GodId,c,{0=>GodId::Lamplighter,1=>GodId::Tollkeeper,2=>GodId::Choir,3=>GodId::Rustmother,4=>GodId::Audit});
tagged!(condition,Condition,c,{0=>Condition::Sound,1=>Condition::Rust,2=>Condition::Damp,3=>Condition::Cracked,4=>Condition::Blessed,5=>Condition::Humming,6=>Condition::Owed,7=>Condition::Spent,8=>Condition::Perishable});
tagged!(job,Job,c,{0=>Job::Lamplighter,1=>Job::Miller,2=>Job::Tinker,3=>Job::Lampwright,4=>Job::Warden,5=>Job::Mender,6=>Job::Fence,7=>Job::Auctioneer,8=>Job::Innkeep,9=>Job::CorpseTender,10=>Job::Chaplain,11=>Job::CoachDriver,12=>Job::Beggar});
tagged!(service,Service,c,{1=>Service::Inn,2=>Service::Tinker,3=>Service::Lampwright,4=>Service::Warden,5=>Service::Mender,6=>Service::Fence,7=>Service::Auction});
tagged!(direction,Direction,c,{1=>Direction::North,2=>Direction::South,3=>Direction::West,4=>Direction::East,5=>Direction::NorthWest,6=>Direction::NorthEast,7=>Direction::SouthWest,8=>Direction::SouthEast});
fn mode_from_tag(tag: u8, c: &str) -> Result<Mode, PersistenceError> {
    match tag {
        1 => Ok(Mode::Classic),
        2 => Ok(Mode::Story),
        3 => Ok(Mode::Ironcartographer),
        _ => Err(PersistenceError::Invalid {
            context: format!("{c} unknown tag {tag}"),
        }),
    }
}
fn mode_tag(v: Mode) -> u8 {
    match v {
        Mode::Classic => 1,
        Mode::Story => 2,
        Mode::Ironcartographer => 3,
    }
}
fn service_tag(v: Service) -> u8 {
    v.canonical_tag()
}
fn direction_tag(v: &Direction) -> u8 {
    v.canonical_tag()
}
fn vault_choice(tag: u8) -> Result<Option<VaultChoice>, PersistenceError> {
    match tag {
        0 => Ok(None),
        1 => Ok(Some(VaultChoice::Seal)),
        2 => Ok(Some(VaultChoice::Crown)),
        _ => Err(PersistenceError::Invalid {
            context: format!("state.vault_choice unknown tag {tag}"),
        }),
    }
}

/// Fallible same-directory atomic byte replacement for save/demo/vault writers.
pub fn write_bytes_atomic(path: impl AsRef<Path>, bytes: &[u8]) -> Result<(), std::io::Error> {
    let path = path.as_ref();
    let parent = path
        .parent()
        .filter(|p| !p.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));
    let name = path.file_name().ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "atomic target has no file name",
        )
    })?;
    let mut last = None;
    for attempt in 0..1000u32 {
        let temp = parent.join(format!(
            ".{}.{}.{}.tmp",
            name.to_string_lossy(),
            std::process::id(),
            attempt
        ));
        let open = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temp);
        let mut file = match open {
            Ok(v) => v,
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                last = Some(e);
                continue;
            }
            Err(e) => return Err(e),
        };
        let result = (|| {
            file.write_all(bytes)?;
            file.flush()?;
            file.sync_all()?;
            drop(file);
            std::fs::rename(&temp, path)?;
            if let Ok(dir) = std::fs::File::open(parent) {
                let _ = dir.sync_all();
            }
            Ok(())
        })();
        if result.is_err() {
            let _ = std::fs::remove_file(&temp);
        }
        return result;
    }
    Err(last.unwrap_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::AlreadyExists,
            "could not reserve atomic temp file",
        )
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use hexwork_core::cause::CauseLink;
    use hexwork_core::character;
    use hexwork_core::runtime::{ActorIdentity, ActorRole};

    fn identity(mode: Mode, seed: u64) -> RunIdentity {
        RunIdentity {
            seed,
            mode,
            ancestry: Ancestry::Mudborn,
            calling: Calling::Ratcatcher,
            display_name: "Ada".into(),
            content_hash: "pack".into(),
            lineage_id: LineageId::from_bytes([1; 16]),
            run_id: RunId::new(seed.max(1)).unwrap(),
            initial_lineage: InitialLineage::default(),
        }
    }

    fn player_state(seed: u64, location: RunLocation) -> (State, u64) {
        let mut state = State::new(seed);
        let player = state.spawn_actor_with_identity(
            hexwork_core::spatial::FixedVec3::from_cell_feet([0; 3]).unwrap(),
            ActorIdentity::new("delver", "Ada", ActorRole::Player),
        );
        state.characters.insert(
            player,
            character::new(Ancestry::Mudborn, Calling::Ratcatcher),
        );
        match location {
            RunLocation::Town(town) => state.initialize_town(town).unwrap(),
            RunLocation::Deep(depth) => {
                let mut image = state.to_image().unwrap();
                image.run_location = RunLocation::Deep(depth);
                state = State::from_image(image).unwrap();
            }
            RunLocation::Uninitialized => {}
        }
        (state, player)
    }

    fn test_item(id: u64) -> Instance {
        Instance {
            id,
            template: "item".into(),
            material: 0,
            identified: false,
            cursed: false,
            blessed: false,
            condition: Condition::Sound,
            silvered: false,
            illegal: false,
            tags: Vec::new(),
            fuel: 0,
            wet_ticks: 0,
        }
    }

    fn raw<T: Encode>(value: T) -> Vec<u8> {
        let bytes = bincode::encode_to_vec(value, bincode::config::standard()).unwrap();
        zstd::stream::encode_all(bytes.as_slice(), 1).unwrap()
    }

    fn run_wire(
        identity: &RunIdentity,
        current: &State,
        checkpoint: Option<&StoryCheckpoint>,
        events: &[SessionEvent],
    ) -> RunWire {
        RunWire {
            header: header(RUN_MAGIC, RUN_FORMAT_VERSION),
            identity: identity_to_wire(identity),
            current: hashed_state_to_wire(HashedState::from_state(current).unwrap()).unwrap(),
            story_checkpoint: checkpoint
                .map(story_checkpoint_to_wire)
                .transpose()
                .unwrap(),
            events: events
                .iter()
                .map(event_to_wire)
                .collect::<Result<_, _>>()
                .unwrap(),
        }
    }

    /// A canonical UseTransition event: the operation spans its authored ten
    /// ticks (docs/fluid §4.2), so it ends at `tick + 10`.
    fn transition(tick: u64) -> SessionEvent {
        SessionEvent::action(
            tick,
            tick + 10,
            CommittedAction::UseTransition { transition_id: 7 },
        )
    }

    fn resurrect(pre: u64, post: u64, prefix: u32, checkpoint_hash: u64) -> SessionEvent {
        SessionEvent {
            pre_tick: pre,
            post_tick: post,
            kind: SessionEventKind::StoryResurrect {
                checkpoint_after_event_count: prefix,
                checkpoint_hash,
                dead_state_hash: 11,
                restored_state_hash: 12,
            },
        }
    }

    /// The format version baked into the frozen pre-transfer fixtures. It is a
    /// historical fact about those bytes, deliberately distinct from the
    /// current development marker, which is asserted only through the crate
    /// constants below.
    const FIXTURE_V4_FORMAT_VERSION: u32 = 4;
    const FIXTURE_V3_FORMAT_VERSION: u32 = 3;

    /// SHA-256 pins for the frozen fixtures. These bytes are immutable
    /// historical inputs captured from accepted commit
    /// 7834d52bcb00f2dec1922d1b995e928e2e991f90 (v4) and the accepted v3
    /// encoder at 97edbd8dc95e805bfe4f9167ea09420d2853baa2. The development
    /// format bump demoted them from supported runtime paths to rejection
    /// inputs, so every test that reads them must also prove they are never
    /// rewritten.
    const CLASSIC_V4_RUN_SHA256: &str =
        "467e582827b6f09a05aabaee60ba5bd4e85c373e5345a6a7aeee56dbf5c340c3";
    const CLASSIC_V4_REPLAY_SHA256: &str =
        "af28de1eaebcd9b66ec0bb72596f8bca22e2c5726219066e9e8b8d43ba997115";
    const STORY_V4_RUN_SHA256: &str =
        "1dc214719fecca479a4e868d0da1ea04b974a297dec1370505d1126263fd5af5";
    const V3_RUN_SHA256: &str =
        "ce90fb2c20f992126e4cbcd869430b336be9e90144552c6dd4c3fd3e216cc21e";
    const V3_REPLAY_SHA256: &str =
        "b1b22217daa7924df42b0969d7a8e017a6b06e074b13992fc403b1b7517210ea";

    fn fixture_path(name: &str) -> std::path::PathBuf {
        Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("testfixtures")
            .join(name)
    }

    fn sha256_hex(bytes: &[u8]) -> String {
        use std::fmt::Write as _;
        let digest = Sha256::digest(bytes);
        let mut hex = String::with_capacity(digest.len() * 2);
        for byte in digest {
            write!(hex, "{byte:02x}").expect("write to String");
        }
        hex
    }

    /// Byte preservation: the embedded bytes and the on-disk frozen fixture
    /// both still hash to the pinned constant, so no test path rewrote them.
    fn assert_fixture_pinned(name: &str, bytes: &[u8], expected: &str) {
        assert_eq!(
            sha256_hex(bytes),
            expected,
            "{name}: embedded frozen bytes drifted"
        );
        let on_disk = std::fs::read(fixture_path(name))
            .unwrap_or_else(|error| panic!("read frozen fixture {name}: {error}"));
        assert_eq!(on_disk, bytes, "{name}: on-disk fixture differs from pin");
        assert_eq!(
            sha256_hex(&on_disk),
            expected,
            "{name}: on-disk frozen fixture was rewritten"
        );
    }

    fn decompressed(fixture: &[u8]) -> Vec<u8> {
        zstd::stream::decode_all(fixture).expect("frozen fixture is a zstd envelope")
    }

    fn recompressed(raw: &[u8]) -> Vec<u8> {
        zstd::stream::encode_all(raw, 1).expect("recompress derived mutant")
    }

    /// Decode the bare envelope header of a decompressed record, returning it
    /// with its exact encoded length. The mutants below are then built relative
    /// to that true header boundary rather than a hardcoded offset.
    fn raw_header(raw: &[u8]) -> (EnvelopeHeader, usize) {
        bincode::decode_from_slice_with_context::<DecodeBudget, EnvelopeHeader, _>(
            raw,
            bincode::config::standard().with_limit::<MAX_DECOMPRESSED_BYTES>(),
            DecodeBudget::default(),
        )
        .expect("frozen fixture carries a decodable bounded header")
    }

    /// AC.1: the frozen pre-transfer v4 run/replay fixtures are rejection
    /// inputs, not supported runtime paths. The bounded header gate refuses
    /// every one of them as `OlderVersion { found: 4, current: <marker> }`
    /// before any payload decode, and the frozen bytes stay pinned.
    #[test]
    fn v4_fixtures_reject_before_body_and_bytes_stay_pinned() {
        let classic = include_bytes!("../testfixtures/run-v4-7834d52-classic.hexrun");
        let story = include_bytes!("../testfixtures/run-v4-7834d52-story.hexrun");
        let replay = include_bytes!("../testfixtures/replay-v4-7834d52-classic.hexdemo");

        assert_fixture_pinned(
            "run-v4-7834d52-classic.hexrun",
            classic,
            CLASSIC_V4_RUN_SHA256,
        );
        assert_fixture_pinned("run-v4-7834d52-story.hexrun", story, STORY_V4_RUN_SHA256);
        assert_fixture_pinned(
            "replay-v4-7834d52-classic.hexdemo",
            replay,
            CLASSIC_V4_REPLAY_SHA256,
        );

        // The fixtures carry the historical version, not the current marker.
        assert!(FIXTURE_V4_FORMAT_VERSION < RUN_FORMAT_VERSION);
        assert!(FIXTURE_V4_FORMAT_VERSION < REPLAY_FORMAT_VERSION);

        assert!(matches!(
            decode_run(classic, "pack"),
            Err(PersistenceError::OlderVersion {
                format: "run-save",
                found: FIXTURE_V4_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
            })
        ));
        assert!(matches!(
            decode_run(story, "pack"),
            Err(PersistenceError::OlderVersion {
                format: "run-save",
                found: FIXTURE_V4_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
            })
        ));
        assert!(matches!(
            decode_replay(replay, "pack"),
            Err(PersistenceError::OlderVersion {
                format: "replay",
                found: FIXTURE_V4_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
            })
        ));
        // A v4 replay body is not a v4 run body: the run gate names its format.
        assert!(matches!(
            decode_run(replay, "pack"),
            Err(PersistenceError::WrongMagic { format: "run-save" })
        ));

        // The gate fires on the raw header, so even a truncated, garbage, or
        // extended body is refused as OlderVersion rather than Decode,
        // CoreImage, or hash mismatch.
        let run_raw = decompressed(classic);
        let (run_header, run_header_len) = raw_header(&run_raw);
        assert_eq!(run_header.magic, RUN_MAGIC);
        assert_eq!(run_header.format_version, FIXTURE_V4_FORMAT_VERSION);
        assert_ne!(run_header.format_version, RUN_FORMAT_VERSION);

        let truncated = run_raw[..run_header_len].to_vec();
        assert!(matches!(
            decode_run(&recompressed(&truncated), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V4_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
                ..
            })
        ));

        let mut garbage = run_raw.clone();
        for (index, byte) in garbage.iter_mut().enumerate().skip(run_header_len) {
            *byte = (index as u8).wrapping_mul(37).wrapping_add(11);
        }
        assert!(matches!(
            decode_run(&recompressed(&garbage), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V4_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
                ..
            })
        ));

        let mut trailing = run_raw.clone();
        trailing.extend_from_slice(&[0xab; 8]);
        assert!(matches!(
            decode_run(&recompressed(&trailing), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V4_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
                ..
            })
        ));

        let mut wrong_magic = run_raw.clone();
        wrong_magic[..8].copy_from_slice(b"HXWRUN99");
        assert!(matches!(
            decode_run(&recompressed(&wrong_magic), "pack"),
            Err(PersistenceError::WrongMagic { format: "run-save" })
        ));

        let replay_raw = decompressed(replay);
        let (replay_header, replay_header_len) = raw_header(&replay_raw);
        assert_eq!(replay_header.magic, REPLAY_MAGIC);
        assert_eq!(replay_header.format_version, FIXTURE_V4_FORMAT_VERSION);

        let replay_truncated = replay_raw[..replay_header_len].to_vec();
        assert!(matches!(
            decode_replay(&recompressed(&replay_truncated), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V4_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
                ..
            })
        ));

        let mut replay_garbage = replay_raw.clone();
        for (index, byte) in replay_garbage.iter_mut().enumerate().skip(replay_header_len) {
            *byte = (index as u8).wrapping_mul(53).wrapping_add(7);
        }
        assert!(matches!(
            decode_replay(&recompressed(&replay_garbage), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V4_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
                ..
            })
        ));

        let mut replay_trailing = replay_raw.clone();
        replay_trailing.extend_from_slice(&[0xcd; 8]);
        assert!(matches!(
            decode_replay(&recompressed(&replay_trailing), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V4_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
                ..
            })
        ));

        let mut replay_wrong_magic = replay_raw.clone();
        replay_wrong_magic[..8].copy_from_slice(b"HXDEMO99");
        assert!(matches!(
            decode_replay(&recompressed(&replay_wrong_magic), "pack"),
            Err(PersistenceError::WrongMagic { format: "replay" })
        ));

        // Nothing above touched the frozen inputs.
        assert_fixture_pinned(
            "run-v4-7834d52-classic.hexrun",
            classic,
            CLASSIC_V4_RUN_SHA256,
        );
        assert_fixture_pinned("run-v4-7834d52-story.hexrun", story, STORY_V4_RUN_SHA256);
        assert_fixture_pinned(
            "replay-v4-7834d52-classic.hexdemo",
            replay,
            CLASSIC_V4_REPLAY_SHA256,
        );
    }

    /// The v3 predecessors are likewise rejection inputs, not supported runtime
    /// paths. These exact files were emitted and decoded by the accepted v3
    /// encoder at 97edbd8dc95e805bfe4f9167ea09420d2853baa2 in an isolated git
    /// archive; the retained bytes are pinned and never regenerated. The header
    /// gate refuses them before the v4 body is even considered.
    #[test]
    fn v3_fixtures_reject_as_older_and_bytes_stay_pinned() {
        let run = include_bytes!("../testfixtures/run-v3-97edbd8.hexrun");
        let replay = include_bytes!("../testfixtures/replay-v3-97edbd8.hexdemo");
        assert_fixture_pinned("run-v3-97edbd8.hexrun", run, V3_RUN_SHA256);
        assert_fixture_pinned("replay-v3-97edbd8.hexdemo", replay, V3_REPLAY_SHA256);

        assert!(FIXTURE_V3_FORMAT_VERSION < RUN_FORMAT_VERSION);
        assert!(FIXTURE_V3_FORMAT_VERSION < REPLAY_FORMAT_VERSION);

        assert!(matches!(
            decode_run(run, "pack"),
            Err(PersistenceError::OlderVersion {
                format: "run-save",
                found: FIXTURE_V3_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
            })
        ));
        assert!(matches!(
            decode_replay(replay, "pack"),
            Err(PersistenceError::OlderVersion {
                format: "replay",
                found: FIXTURE_V3_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
            })
        ));

        // The v3 headers really bake version 3, and their bodies are never
        // interpreted: a truncated body, a garbage body, trailing bytes, and a
        // foreign magic all resolve before payload parsing runs.
        let run_raw = decompressed(run);
        let (run_header, run_header_len) = raw_header(&run_raw);
        assert_eq!(run_header.magic, RUN_MAGIC);
        assert_eq!(run_header.format_version, FIXTURE_V3_FORMAT_VERSION);
        let run_truncated = run_raw[..run_header_len].to_vec();
        assert!(matches!(
            decode_run(&recompressed(&run_truncated), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V3_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
                ..
            })
        ));
        let mut run_garbage = run_raw.clone();
        for (index, byte) in run_garbage.iter_mut().enumerate().skip(run_header_len) {
            *byte = (index as u8).wrapping_mul(17).wrapping_add(29);
        }
        assert!(matches!(
            decode_run(&recompressed(&run_garbage), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V3_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
                ..
            })
        ));
        let mut run_trailing = run_raw.clone();
        run_trailing.extend_from_slice(&[0x5a; 5]);
        assert!(matches!(
            decode_run(&recompressed(&run_trailing), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V3_FORMAT_VERSION,
                current: RUN_FORMAT_VERSION,
                ..
            })
        ));
        let mut run_wrong_magic = run_raw.clone();
        run_wrong_magic[..8].copy_from_slice(b"HXWRUN01");
        assert!(matches!(
            decode_run(&recompressed(&run_wrong_magic), "pack"),
            Err(PersistenceError::WrongMagic { format: "run-save" })
        ));

        let replay_raw = decompressed(replay);
        let (replay_header, replay_header_len) = raw_header(&replay_raw);
        assert_eq!(replay_header.magic, REPLAY_MAGIC);
        assert_eq!(replay_header.format_version, FIXTURE_V3_FORMAT_VERSION);
        let replay_truncated = replay_raw[..replay_header_len].to_vec();
        assert!(matches!(
            decode_replay(&recompressed(&replay_truncated), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V3_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
                ..
            })
        ));
        let mut replay_garbage = replay_raw.clone();
        for (index, byte) in replay_garbage.iter_mut().enumerate().skip(replay_header_len) {
            *byte = (index as u8).wrapping_mul(31).wrapping_add(3);
        }
        assert!(matches!(
            decode_replay(&recompressed(&replay_garbage), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V3_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
                ..
            })
        ));
        let mut replay_trailing = replay_raw.clone();
        replay_trailing.extend_from_slice(&[0x6b; 5]);
        assert!(matches!(
            decode_replay(&recompressed(&replay_trailing), "pack"),
            Err(PersistenceError::OlderVersion {
                found: FIXTURE_V3_FORMAT_VERSION,
                current: REPLAY_FORMAT_VERSION,
                ..
            })
        ));
        let mut replay_wrong_magic = replay_raw.clone();
        replay_wrong_magic[..8].copy_from_slice(b"HXDEMO00");
        assert!(matches!(
            decode_replay(&recompressed(&replay_wrong_magic), "pack"),
            Err(PersistenceError::WrongMagic { format: "replay" })
        ));

        // Nothing above touched the frozen inputs.
        assert_fixture_pinned("run-v3-97edbd8.hexrun", run, V3_RUN_SHA256);
        assert_fixture_pinned("replay-v3-97edbd8.hexdemo", replay, V3_REPLAY_SHA256);
    }

    #[test]
    fn current_and_story_checkpoint_knowledge_require_strict_source_order() {
        fn set_rows(knowledge: &mut KnowledgeWire, category: usize, rows: &[&str]) {
            match category {
                0 => {
                    knowledge.mapped_regions = rows
                        .iter()
                        .map(|row| KnowledgeString::from((*row).to_owned()))
                        .collect()
                }
                1 => {
                    knowledge.recipes = rows
                        .iter()
                        .map(|row| KnowledgeString::from((*row).to_owned()))
                        .collect()
                }
                2 => {
                    knowledge.translated_hymns = rows
                        .iter()
                        .map(|row| KnowledgeString::from((*row).to_owned()))
                        .collect()
                }
                _ => unreachable!(),
            }
        }
        fn assert_ordering_error(error: PersistenceError, category: &str) {
            assert!(
                matches!(error, PersistenceError::Invalid { ref context }
                    if context.contains(category)
                        && context.contains("strictly increasing and unique")
                        && !context.contains("hash")),
                "unexpected error: {error}"
            );
        }

        let seed = 6;
        let (current, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&current).unwrap(),
            after_event_count: 0,
        };
        for (category_index, category_name) in [
            (0, "mapped_regions"),
            (1, "recipes"),
            (2, "translated_hymns"),
        ] {
            for rows in [["alpha", "alpha"], ["bravo", "alpha"]] {
                let mut current_wire = run_wire(
                    &identity(Mode::Story, seed),
                    &current,
                    Some(&checkpoint),
                    &[],
                );
                set_rows(
                    &mut current_wire.current.image.knowledge,
                    category_index,
                    &rows,
                );
                assert_ordering_error(
                    decode_run(&raw(current_wire), "pack").unwrap_err(),
                    category_name,
                );

                let mut checkpoint_wire = run_wire(
                    &identity(Mode::Story, seed),
                    &current,
                    Some(&checkpoint),
                    &[],
                );
                set_rows(
                    &mut checkpoint_wire
                        .story_checkpoint
                        .as_mut()
                        .unwrap()
                        .state
                        .image
                        .knowledge,
                    category_index,
                    &rows,
                );
                assert_ordering_error(
                    decode_run(&raw(checkpoint_wire), "pack").unwrap_err(),
                    category_name,
                );
            }
        }
    }

    #[test]
    fn v4_event_tags_payloads_and_caps_are_canonical() {
        let action = SessionEvent::action(0, 1, CommittedAction::Wait);
        let action_wire = event_to_wire(&action).unwrap();
        assert_eq!(action_wire.kind_tag, 1);
        assert_eq!(action_wire.action.tag, 2);
        assert_eq!(action_wire.checkpoint_hash, 0);
        let life = resurrect(1, 0, 0, 5);
        let life_wire = event_to_wire(&life).unwrap();
        assert_eq!(life_wire.kind_tag, 2);
        assert_eq!(life_wire.action.tag, 0);
        assert_eq!(life_wire.checkpoint_hash, 5);

        let mut bad_action = action_wire.clone();
        bad_action.dead_state_hash = 1;
        assert!(wire_to_event(bad_action, 0, "events").is_err());
        let mut bad_life = life_wire.clone();
        bad_life.action.tag = 2;
        assert!(wire_to_event(bad_life, 0, "events").is_err());
        let mut unknown = life_wire;
        unknown.kind_tag = 99;
        assert!(wire_to_event(unknown, 0, "events").is_err());

        let prefix =
            bincode::encode_to_vec((MAX_SESSION_EVENTS + 1) as u64, bincode::config::standard())
                .unwrap();
        let decoded = bincode::decode_from_slice_with_context::<
            DecodeBudget,
            BoundedVec<SessionEventWire, MAX_SESSION_EVENTS, BUDGET_EVENTS>,
            _,
        >(
            &prefix,
            bincode::config::standard(),
            DecodeBudget::default(),
        );
        assert!(decoded
            .err()
            .unwrap()
            .to_string()
            .contains("per-field limit"));
    }

    #[test]
    fn tinker_tag12_canonical_roundtrip() {
        for item_id in [1, 41, u64::MAX] {
            let action = CommittedAction::TinkerIdentify { item_id };
            let wire = action_to_wire(&action).unwrap();
            assert_eq!(
                (wire.tag, wire.id, wire.direction, wire.service),
                (12, item_id, 0, 0)
            );
            assert!(wire.text.is_empty());
            assert_eq!(wire_to_action(wire, "action").unwrap(), action);
            let event = SessionEvent::action(0, 20, action);
            assert_eq!(
                wire_to_event(event_to_wire(&event).unwrap(), 0, "events").unwrap(),
                event
            );
            // Synthetic codec record, not gameplay ownership or acquisition evidence.
            let (mut state, _) = player_state(42, RunLocation::Deep(1));
            state.tick = 20;
            let id = identity(Mode::Classic, 42);
            let events = vec![event];
            let bytes = encode_run(&id, &state, None, &events).unwrap();
            assert_eq!(decode_run(&bytes, "pack").unwrap().events, events);
            let replay = ProductionReplay {
                identity: id,
                events,
                expected_final_tick: 20,
                expected_final_hash: state.try_hash_state().unwrap(),
            };
            assert_eq!(
                decode_replay(&encode_replay(&replay).unwrap(), "pack").unwrap(),
                replay
            );
        }
    }

    #[test]
    fn tinker_tag12_rejects_unused_payload() {
        assert!(action_to_wire(&CommittedAction::TinkerIdentify { item_id: 0 }).is_err());
        let wire = action_to_wire(&CommittedAction::TinkerIdentify { item_id: 1 }).unwrap();
        for mutation in 0..5 {
            let mut invalid = wire.clone();
            match mutation {
                0 => invalid.id = 0,
                1 => invalid.direction = 1,
                2 => invalid.service = 1,
                3 => invalid.text = "unused".to_owned().into(),
                _ => invalid.tag = 13,
            }
            assert!(wire_to_action(invalid, "action").is_err());
        }
    }

    /// Every committed operation now carries its real authored span
    /// (docs/fluid §4.2): Move is one full tick, InspectBoard is observational
    /// (zero), and UseTransition costs ten. A wrong span is rejected.
    #[test]
    fn move_action_costs_exactly_one_tick_and_rejects_zero_span() {
        assert_eq!(
            action_semantic_cost(&CommittedAction::Move(Direction::South)),
            Ok((1, 0))
        );
        for mode in [Mode::Classic, Mode::Story] {
            assert!(
                timeline_costs(
                    &[SessionEvent::action(0, 0, CommittedAction::Move(Direction::North))],
                    mode,
                    0
                )
                .is_err(),
                "a zero-tick Move must be rejected"
            );
            assert_eq!(
                timeline_costs(
                    &[SessionEvent::action(0, 1, CommittedAction::Move(Direction::North))],
                    mode,
                    1
                )
                .unwrap(),
                TimelineCosts {
                    full_steps: 1,
                    unattended_steps: 0,
                    total_elapsed_ticks: 1,
                }
            );
        }
        // InspectBoard is observational; UseTransition costs its authored ten.
        assert_eq!(
            action_semantic_cost(&CommittedAction::InspectBoard {
                town: "milthollow".into()
            }),
            Ok((0, 0))
        );
        assert_eq!(
            action_semantic_cost(&CommittedAction::UseTransition { transition_id: 1 }),
            Ok((10, 0))
        );
        // A ten-tick action with span ten validates; span zero rejects.
        assert_eq!(
            timeline_costs(
                &[SessionEvent::action(0, 10, CommittedAction::Pickup { item_id: 1 })],
                Mode::Classic,
                10
            )
            .unwrap(),
            TimelineCosts {
                full_steps: 10,
                unattended_steps: 0,
                total_elapsed_ticks: 10,
            }
        );
        assert!(
            timeline_costs(
                &[SessionEvent::action(0, 0, CommittedAction::Pickup { item_id: 1 })],
                Mode::Classic,
                0
            )
            .is_err(),
            "a ten-tick action with zero span must be rejected"
        );
    }

    /// TinkerIdentify is an authored twenty-tick world action; the earlier
    /// zero-tick interim is gone. A wrong span is rejected.
    #[test]
    fn tinker_identify_costs_twenty_ticks() {
        let action = CommittedAction::TinkerIdentify { item_id: 41 };
        for mode in [Mode::Classic, Mode::Story] {
            assert_eq!(
                timeline_costs(&[SessionEvent::action(0, 20, action.clone())], mode, 20).unwrap(),
                TimelineCosts {
                    full_steps: 20,
                    unattended_steps: 0,
                    total_elapsed_ticks: 20,
                }
            );
            assert!(
                timeline_costs(&[SessionEvent::action(0, 19, action.clone())], mode, 19).is_err()
            );
            let events = [
                SessionEvent::action(0, 1, CommittedAction::Wait),
                SessionEvent::action(1, 21, action.clone()),
            ];
            assert!(timeline_costs(&events, mode, 21).is_ok());
        }
        // Every non-Inn service is an authored ten-tick world action.
        assert_eq!(
            action_semantic_cost(&CommittedAction::UseService {
                service: Service::Tinker
            }),
            Ok((10, 0))
        );
    }

    /// Every committed operation's journal span equals the core `action_timing`
    /// table — exhaustive over `CommittedAction`, no default arm. Save must not
    /// re-derive timing from a hand-written subset.
    #[test]
    fn every_committed_action_cost_matches_core_action_timing() {
        let inn = Service::Inn;
        let actions = [
            CommittedAction::Move(Direction::North),
            CommittedAction::Wait,
            CommittedAction::InspectBoard {
                town: "milthollow".into(),
            },
            CommittedAction::AcceptContract { id: "rat-1".into() },
            CommittedAction::UseTransition { transition_id: 7 },
            CommittedAction::ResolveAttack { target_actor: 8 },
            CommittedAction::Pickup { item_id: 41 },
            CommittedAction::Drop { item_id: 42 },
            CommittedAction::Wield { item_id: 43 },
            CommittedAction::TinkerIdentify { item_id: 44 },
            CommittedAction::TurnIn {
                contract_id: "rat-1".into(),
            },
            CommittedAction::UseService { service: inn },
            CommittedAction::UseService {
                service: Service::Lampwright,
            },
            CommittedAction::UseService {
                service: Service::Mender,
            },
        ];
        for action in actions {
            let expected = match hexwork_core::action_timing::action_timing(&action) {
                hexwork_core::action_timing::ActionTiming::Observation => (0, 0),
                hexwork_core::action_timing::ActionTiming::World(ticks) => (ticks.get(), 0),
                hexwork_core::action_timing::ActionTiming::Unattended(ticks) => (0, ticks.get()),
            };
            assert_eq!(
                action_semantic_cost(&action).unwrap(),
                expected,
                "{action:?}"
            );
        }
    }

    /// The pending slot round-trips through the run wire, charges the shared
    /// state-row budget, and rejects a duplicate second slot row.
    #[test]
    fn pending_slot_wire_roundtrip_budget_and_duplicate_rejection() {
        let seed = 71;
        let (mut state, _) = player_state(seed, RunLocation::Deep(1));
        state.tick = 30;
        state.pending = Some(Pending::admit(&CommittedAction::Wait, 30));
        let id = identity(Mode::Classic, seed);
        let bytes = encode_run(&id, &state, None, &[]).unwrap();
        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.current.pending, state.pending);

        // The resumable Inn debt survives a run save exactly (no re-debit).
        state.pending = Some(Pending {
            start_tick: 0,
            remaining_ticks: 5_000,
            action: PendingAction::Inn(PendingInn {
                total_ticks: 8_640,
                remaining_ticks: 5_000,
            }),
        });
        let bytes = encode_run(&id, &state, None, &[]).unwrap();
        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.current.pending, state.pending);
        assert_eq!(
            decoded
                .current
                .pending
                .as_ref()
                .unwrap()
                .inn()
                .unwrap()
                .remaining_ticks,
            5_000
        );
        assert_eq!(
            decoded.current.try_hash_state().unwrap(),
            state.try_hash_state().unwrap()
        );

        type PendingVec = BoundedVec<PendingRowWire, MAX_PENDING_ROWS, BUDGET_STATE_ROWS>;
        let row = pending_row_to_wire(state.pending.as_ref().unwrap()).unwrap();
        let one = bincode::encode_to_vec(
            PendingVec::from(vec![row.clone()]),
            bincode::config::standard(),
        )
        .unwrap();
        let decoded_one = bincode::decode_from_slice_with_context::<DecodeBudget, PendingVec, _>(
            &one,
            bincode::config::standard(),
            DecodeBudget::default(),
        )
        .unwrap()
        .0;
        assert_eq!(decoded_one.len(), 1);

        // Two rows for the single slot is a duplicate-slot violation.
        let two = bincode::encode_to_vec(
            PendingVec::from(vec![row.clone(), row]),
            bincode::config::standard(),
        )
        .unwrap();
        assert!(bincode::decode_from_slice_with_context::<DecodeBudget, PendingVec, _>(
            &two,
            bincode::config::standard(),
            DecodeBudget::default(),
        )
        .err()
        .unwrap()
        .to_string()
        .contains("per-field limit"));

        // A pending row charges the shared total-state-rows aggregate.
        let saturated = DecodeBudget {
            state_rows: MAX_TOTAL_STATE_ROWS,
            ..DecodeBudget::default()
        };
        assert!(bincode::decode_from_slice_with_context::<DecodeBudget, PendingVec, _>(
            &one,
            bincode::config::standard(),
            saturated,
        )
        .err()
        .unwrap()
        .to_string()
        .contains("total state rows"));
    }

    /// Observation rows round-trip through the run wire, charge the shared
    /// total-state-rows aggregate, and are rejected when duplicated or carrying
    /// an unknown face tag.
    #[test]
    fn observation_rows_wire_roundtrip_budget_and_duplicate_rejection() {
        let seed = 71;
        let mut state = State::new(seed);
        for y in -2..=2 {
            for z in 0..=3 {
                state.grid.set_world(
                    2,
                    y,
                    z,
                    hexwork_core::voxel::Voxel {
                        material: hexwork_core::voxel::material::STONE,
                        occ: 1,
                        ..Default::default()
                    },
                );
            }
        }
        for x in 0..=1 {
            for y in -3..=3 {
                for z in 0..=3 {
                    let voxel = state.grid.get_world(x, y, z);
                    if voxel.occ == 0 {
                        state.grid.set_world(
                            x,
                            y,
                            z,
                            hexwork_core::voxel::Voxel { light: 8, ..voxel },
                        );
                    }
                }
            }
        }
        let player = state.spawn_actor_with_identity(
            hexwork_core::spatial::FixedVec3::from_cell_feet([0; 3]).unwrap(),
            ActorIdentity::new("delver", "Ada", ActorRole::Player),
        );
        state.characters.insert(
            player,
            character::new(Ancestry::Mudborn, Calling::Ratcatcher),
        );
        state.initialize_town("milthollow").unwrap();
        assert!(!state.observations().is_empty());

        let id = identity(Mode::Classic, seed);
        let bytes = encode_run(&id, &state, None, &[]).unwrap();
        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.current.observations(), state.observations());
        assert_eq!(decoded.current.observation_rows(), state.observation_rows());
        assert_eq!(
            decoded.current.try_hash_state().unwrap(),
            state.try_hash_state().unwrap()
        );

        // Observation rows charge the shared total-state-rows aggregate.
        type ObsVec = BoundedVec<ObservationRowWire, MAX_OBSERVATION_ROWS, BUDGET_STATE_ROWS>;
        let wire_rows: Vec<ObservationRowWire> = state
            .observation_rows()
            .iter()
            .map(observation_row_to_wire)
            .collect::<Result<_, _>>()
            .unwrap();
        let encoded =
            bincode::encode_to_vec(ObsVec::from(wire_rows), bincode::config::standard()).unwrap();
        let saturated = DecodeBudget {
            state_rows: MAX_TOTAL_STATE_ROWS,
            ..DecodeBudget::default()
        };
        assert!(bincode::decode_from_slice_with_context::<DecodeBudget, ObsVec, _>(
            &encoded,
            bincode::config::standard(),
            saturated,
        )
        .err()
        .unwrap()
        .to_string()
        .contains("total state rows"));

        // A duplicated row is a non-canonical row vector, rejected on decode.
        let mut wire = run_wire(&id, &state, None, &[]);
        let duplicate = wire.current.image.observations[0].clone();
        wire.current.image.observations.0.push(duplicate);
        let err = decode_run(&raw(wire), "pack").unwrap_err();
        assert!(
            err.to_string().contains("duplicate or noncanonical order"),
            "unexpected duplicate error: {err}"
        );

        // An unknown face tag is rejected rather than defaulted.
        let mut wire = run_wire(&id, &state, None, &[]);
        wire.current.image.observations[0].face = 9;
        assert!(decode_run(&raw(wire), "pack").is_err());
    }

    /// The durable branch epoch round-trips through the run wire and adds no
    /// rows to the shared state-row budget.
    #[test]
    fn branch_epoch_wire_roundtrip_and_budget_rows() {
        let seed = 73;
        let id = identity(Mode::Classic, seed);

        let (mut state, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        state.branch_epoch = 3;
        let bytes = encode_run(&id, &state, None, &[]).unwrap();
        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.current.branch_epoch, 3);
        assert_eq!(
            decoded.current.try_hash_state().unwrap(),
            state.try_hash_state().unwrap()
        );

        let wire = image_to_wire(state.to_image().unwrap()).unwrap();
        assert_eq!(wire.branch_epoch, 3);
        assert_eq!(wire_to_image(wire).unwrap().branch_epoch, 3);

        // The epoch is a scalar, so it adds no rows to the shared budget while
        // still changing the canonical hash.
        let (mut baseline, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        baseline.branch_epoch = 0;
        assert_eq!(
            baseline.observation_rows().len(),
            state.observation_rows().len(),
            "the branch epoch adds no observation rows"
        );
        assert_ne!(
            baseline.try_hash_state().unwrap(),
            state.try_hash_state().unwrap(),
            "the branch epoch still changes the canonical hash"
        );
    }

    #[test]
    fn wield_v4_tag11_is_canonical_and_old_bytes_stay_stable() {
        for item_id in [1, 41, u64::MAX] {
            let action = CommittedAction::Wield { item_id };
            let wire = action_to_wire(&action).unwrap();
            assert_eq!(
                (wire.tag, wire.id, wire.direction, wire.service),
                (11, item_id, 0, 0)
            );
            assert!(wire.text.is_empty());
            assert_eq!(wire_to_action(wire.clone(), "action").unwrap(), action);
            for mutation in 0..5 {
                let mut invalid = wire.clone();
                match mutation {
                    0 => invalid.id = 0,
                    1 => invalid.direction = 1,
                    2 => invalid.service = 1,
                    3 => invalid.text = "unused".to_owned().into(),
                    _ => invalid.tag = 13,
                }
                assert!(wire_to_action(invalid, "action").is_err());
            }
            // The authored duration is the single source of truth (docs/fluid
            // §4.2, `action_timing`): a canonical Wield spans exactly its
            // authored world ticks, so the span-0 record is no longer canonical
            // and both too-short and too-long spans are refused.
            let (world_ticks, unattended_ticks) = action_semantic_cost(&action).unwrap();
            assert_eq!(
                (world_ticks, unattended_ticks),
                (10, 0),
                "Wield is authored as ten world ticks"
            );
            let span = world_ticks + unattended_ticks;
            let event = SessionEvent::action(0, span, action.clone());
            assert_eq!(
                wire_to_event(event_to_wire(&event).unwrap(), 0, "events").unwrap(),
                event
            );
            for mode in [Mode::Classic, Mode::Story] {
                assert_eq!(
                    timeline_costs(std::slice::from_ref(&event), mode, span).unwrap(),
                    TimelineCosts {
                        full_steps: world_ticks,
                        unattended_steps: unattended_ticks,
                        total_elapsed_ticks: span,
                    }
                );
                assert!(
                    timeline_costs(&[SessionEvent::action(0, 1, action.clone())], mode, 1).is_err(),
                    "a Wield shorter than its authored span must be rejected"
                );
                assert!(
                    timeline_costs(&[SessionEvent::action(0, span + 1, action.clone())], mode, span)
                        .is_err(),
                    "a Wield longer than its authored span must be rejected"
                );
            }
        }
        assert!(action_to_wire(&CommittedAction::Wield { item_id: 0 }).is_err());
        let id = identity(Mode::Classic, 42);
        let (mut state, _) = player_state(42, RunLocation::Deep(1));
        /// One canonical ten-tick Wield starting at `index * 10`.
        fn wield(index: u64) -> SessionEvent {
            SessionEvent::action(
                index * 10,
                index * 10 + 10,
                CommittedAction::Wield { item_id: 1 },
            )
        }
        // A canonical ten-tick Wield ends at tick ten, so the state it is
        // retained alongside carries tick ten.
        state.tick = 10;
        let image = state.to_image().unwrap();
        let mut full: Vec<SessionEvent> = (0..MAX_SESSION_EVENTS as u64).map(wield).collect();
        let next = wield(MAX_SESSION_EVENTS as u64);
        let expected = PersistenceError::Limit {
            context: "run.events".into(),
            actual: MAX_SESSION_EVENTS + 1,
            limit: MAX_SESSION_EVENTS,
        };
        assert_eq!(
            validate_run_structural_limits(&id, &image, None, &full, Some(&next)),
            Err(expected.clone())
        );
        full.push(next);
        assert_eq!(encode_run(&id, &state, None, &full), Err(expected));
        // Synthetic codec record: gameplay ownership/replay execution is tested by C/P.
        let events = vec![wield(0)];
        let bytes = encode_run(&id, &state, None, &events).unwrap();
        assert_eq!(decode_run(&bytes, "pack").unwrap().events, events);
        let replay = ProductionReplay {
            identity: id,
            events,
            expected_final_tick: 10,
            expected_final_hash: state.try_hash_state().unwrap(),
        };
        assert_eq!(
            decode_replay(&encode_replay(&replay).unwrap(), "pack").unwrap(),
            replay
        );
        v4_fixtures_reject_before_body_and_bytes_stay_pinned();
        v3_fixtures_reject_as_older_and_bytes_stay_pinned();
    }

    #[test]
    fn transfer_tags_and_timing_are_canonical() {
        let old_actions = [
            (CommittedAction::Move(Direction::North), 1),
            (CommittedAction::Wait, 2),
            (
                CommittedAction::InspectBoard {
                    town: "milthollow".into(),
                },
                3,
            ),
            (CommittedAction::AcceptContract { id: "rat-1".into() }, 4),
            (CommittedAction::UseTransition { transition_id: 7 }, 5),
            (CommittedAction::ResolveAttack { target_actor: 8 }, 6),
            (
                CommittedAction::TurnIn {
                    contract_id: "rat-1".into(),
                },
                7,
            ),
            (
                CommittedAction::UseService {
                    service: Service::Inn,
                },
                8,
            ),
        ];
        for (action, tag) in old_actions {
            assert_eq!(action_to_wire(&action).unwrap().tag, tag);
        }

        for (action, tag, id) in [
            (CommittedAction::Pickup { item_id: 41 }, 9, 41),
            (CommittedAction::Drop { item_id: 42 }, 10, 42),
        ] {
            let wire = action_to_wire(&action).unwrap();
            assert_eq!(wire.tag, tag);
            assert_eq!(wire.id, id);
            assert_eq!(wire.direction, 0);
            assert_eq!(wire.service, 0);
            assert!(wire.text.is_empty());
            assert_eq!(wire_to_action(wire, "action").unwrap(), action);
            assert_eq!(
                timeline_costs(
                    &[SessionEvent::action(0, 10, action)],
                    Mode::Classic,
                    10
                )
                .unwrap(),
                TimelineCosts {
                    full_steps: 10,
                    unattended_steps: 0,
                    total_elapsed_ticks: 10,
                }
            );
        }
        for action in [
            CommittedAction::Pickup { item_id: 0 },
            CommittedAction::Drop { item_id: 0 },
        ] {
            assert!(action_to_wire(&action).is_err());
        }
        for tag in [9, 10] {
            for mutate in 0..5 {
                let mut wire = ActionWire {
                    tag,
                    direction: 0,
                    service: 0,
                    text: BoundedString::default(),
                    id: 1,
                };
                match mutate {
                    0 => wire.id = 0,
                    1 => wire.direction = 1,
                    2 => wire.service = 1,
                    3 => wire.text = "x".to_owned().into(),
                    _ => wire.tag = 13,
                }
                assert!(wire_to_action(wire, "action").is_err());
            }
        }
        assert!(timeline_costs(
            &[SessionEvent::action(
                0,
                1,
                CommittedAction::Pickup { item_id: 1 }
            )],
            Mode::Classic,
            1,
        )
        .is_err());
    }

    #[test]
    fn event_view_empty_base_append_indexing_and_order_match_materialized() {
        let first = transition(0);
        let second = SessionEvent::action(0, 1, CommittedAction::Wait);
        let appended = SessionEvent::action(1, 1, CommittedAction::Pickup { item_id: 9 });

        let empty = EventView::new(&[], None).unwrap();
        assert_eq!(empty.checked_len().unwrap(), 0);
        assert!(empty.get(0).is_none());
        assert_eq!(empty.iter().count(), 0);

        let base = [first, second];
        let base_only = EventView::new(&base, None).unwrap();
        assert_eq!(base_only.checked_len().unwrap(), 2);
        assert_eq!(base_only.get(0), Some(&base[0]));
        assert_eq!(base_only.get(1), Some(&base[1]));
        assert!(base_only.get(2).is_none());

        let prospective = EventView::new(&base, Some(&appended)).unwrap();
        assert_eq!(prospective.checked_len().unwrap(), 3);
        assert_eq!(prospective.get(2), Some(&appended));
        assert!(prospective.get(3).is_none());
        let materialized = base
            .iter()
            .chain(std::iter::once(&appended))
            .collect::<Vec<_>>();
        assert_eq!(prospective.iter().collect::<Vec<_>>(), materialized);
    }

    #[test]
    fn resurrection_tag_cannot_decode_as_player_action() {
        let wire = event_to_wire(&resurrect(0, 0, 0, 1)).unwrap();
        assert_eq!(wire.kind_tag, 2);
        assert!(wire_to_action(wire.action, "action").is_err());
    }

    #[test]
    fn history_across_rollback_survives_roundtrip() {
        let replay = ProductionReplay {
            identity: identity(Mode::Story, 7),
            events: vec![
                transition(0),
                resurrect(10, 10, 1, 44),
                SessionEvent::action(10, 11, CommittedAction::Wait),
                SessionEvent::action(11, 12, CommittedAction::Wait),
            ],
            expected_final_tick: 12,
            expected_final_hash: 99,
        };
        let decoded = decode_replay(&encode_replay(&replay).unwrap(), "pack").unwrap();
        assert_eq!(decoded, replay);
        assert_eq!(decoded.events.len(), 4);
    }

    #[test]
    fn story_deep_current_town_checkpoint_exact_dual_state_hash_and_continuation_roundtrip() {
        let seed = 17;
        let (checkpoint_state, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&checkpoint_state).unwrap(),
            after_event_count: 0,
        };
        let (mut current, _) = player_state(seed, RunLocation::Deep(1));
        current.step();
        let expected_current_hash = current.try_hash_state().unwrap();
        let expected_checkpoint_hash = checkpoint.state.hash;
        let bytes = encode_run(
            &identity(Mode::Story, seed),
            &current,
            Some(&checkpoint),
            &[],
        )
        .unwrap();
        let mut decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(
            decoded.current.try_hash_state().unwrap(),
            expected_current_hash
        );
        assert_eq!(
            decoded.story_checkpoint.as_ref().unwrap().state.hash,
            expected_checkpoint_hash
        );
        assert!(matches!(
            decoded.story_checkpoint.as_ref().unwrap().state.image.run_location,
            RunLocation::Town(ref town) if town == "milthollow"
        ));
        current.step();
        decoded.current.step();
        assert_eq!(
            decoded.current.try_hash_state().unwrap(),
            current.try_hash_state().unwrap()
        );
    }

    #[test]
    fn mode_checkpoint_presence_seed_tick_hash_identity_prefix_rejection() {
        let seed = 19;
        let (town, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&town).unwrap(),
            after_event_count: 0,
        };
        let (deep, _) = player_state(seed, RunLocation::Deep(1));
        assert!(encode_run(&identity(Mode::Story, seed), &deep, None, &[]).is_err());
        assert!(encode_run(
            &identity(Mode::Classic, seed),
            &deep,
            Some(&checkpoint),
            &[]
        )
        .is_err());

        let mut bad = checkpoint.clone();
        bad.state.image.seed += 1;
        assert!(encode_run(&identity(Mode::Story, seed), &deep, Some(&bad), &[]).is_err());
        let mut bad = checkpoint.clone();
        bad.state.tick += 1;
        assert!(encode_run(&identity(Mode::Story, seed), &deep, Some(&bad), &[]).is_err());
        let mut bad = checkpoint.clone();
        bad.state.hash ^= 1;
        assert!(encode_run(&identity(Mode::Story, seed), &deep, Some(&bad), &[]).is_err());

        let mut wrong_name = identity(Mode::Story, seed);
        wrong_name.display_name = "Bea".into();
        assert!(encode_run(&wrong_name, &deep, Some(&checkpoint), &[]).is_err());

        let events = vec![SessionEvent::action(
            0,
            1,
            CommittedAction::Move(Direction::North),
        )];
        let bad_prefix = StoryCheckpoint {
            after_event_count: 1,
            ..checkpoint
        };
        assert!(encode_run(
            &identity(Mode::Story, seed),
            &deep,
            Some(&bad_prefix),
            &events
        )
        .is_err());
    }

    #[test]
    fn current_dead_player_allowed_only_checkpoint_player_alive() {
        let seed = 23;
        let (town, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&town).unwrap(),
            after_event_count: 0,
        };
        let (mut deep, player) = player_state(seed, RunLocation::Deep(1));
        let actor = deep
            .actors
            .iter_mut()
            .find(|actor| actor.id == player)
            .unwrap();
        actor.alive = false;
        actor.hp = 0;
        assert!(encode_run(&identity(Mode::Story, seed), &deep, Some(&checkpoint), &[]).is_ok());

        for mode in [Mode::Classic, Mode::Ironcartographer, Mode::Story] {
            for location in [RunLocation::Town("milthollow".into()), RunLocation::Deep(1)] {
                for (alive, hp) in [(false, 0), (false, -1), (true, 0)] {
                    let (mut current, player) = player_state(seed, location.clone());
                    let actor = current.actors.iter_mut().find(|a| a.id == player).unwrap();
                    actor.alive = alive;
                    actor.hp = hp;
                    let saved_checkpoint = (mode == Mode::Story).then_some(&checkpoint);
                    let bytes = encode_run(&identity(mode, seed), &current, saved_checkpoint, &[])
                        .expect("current dead or pending-death states are valid in Town and Deep");
                    let restored = decode_run(&bytes, "pack").unwrap();
                    assert_eq!(restored.current.run_location(), &location);
                    let actor = restored
                        .current
                        .actors
                        .iter()
                        .find(|a| a.id == player)
                        .unwrap();
                    assert_eq!((actor.alive, actor.hp), (alive, hp));
                    assert_eq!(
                        restored.current.try_hash_state().unwrap(),
                        current.try_hash_state().unwrap()
                    );
                }
            }
        }

        let mut dead_checkpoint = checkpoint;
        let actor = dead_checkpoint
            .state
            .image
            .actors
            .iter_mut()
            .find(|a| a.id == player)
            .unwrap();
        actor.alive = false;
        actor.hp = 0;
        dead_checkpoint = StoryCheckpoint {
            state: HashedState::from_image(dead_checkpoint.state.image).unwrap(),
            after_event_count: 0,
        };
        assert!(encode_run(
            &identity(Mode::Story, seed),
            &deep,
            Some(&dead_checkpoint),
            &[]
        )
        .is_err());
    }

    #[test]
    fn prefix_digest_stability_and_mutation() {
        let base = vec![
            transition(0),
            SessionEvent::action(0, 1, CommittedAction::Wait),
            resurrect(1, 0, 1, 7),
        ];
        let digest = event_prefix_digest(&base, 2).unwrap();
        let mut appended = base.clone();
        appended.push(SessionEvent::action(
            0,
            0,
            CommittedAction::Move(Direction::East),
        ));
        assert_eq!(event_prefix_digest(&appended, 2).unwrap(), digest);
        for mutate in 0..6 {
            let mut changed = base.clone();
            match mutate {
                0 => changed[0].pre_tick = 1,
                1 => changed[1].post_tick = 2,
                2 => {
                    changed[1].kind =
                        SessionEventKind::Action(CommittedAction::Move(Direction::North))
                }
                3 => {
                    if let SessionEventKind::StoryResurrect {
                        ref mut checkpoint_hash,
                        ..
                    } = changed[2].kind
                    {
                        *checkpoint_hash ^= 1
                    }
                }
                4 => {
                    if let SessionEventKind::StoryResurrect {
                        ref mut dead_state_hash,
                        ..
                    } = changed[2].kind
                    {
                        *dead_state_hash ^= 1
                    }
                }
                _ => {
                    if let SessionEventKind::StoryResurrect {
                        ref mut restored_state_hash,
                        ..
                    } = changed[2].kind
                    {
                        *restored_state_hash ^= 1
                    }
                }
            }
            assert_ne!(
                event_prefix_digest(&changed, 3).unwrap(),
                event_prefix_digest(&base, 3).unwrap()
            );
        }
        assert!(event_prefix_digest(&base, 4).is_err());
    }

    #[test]
    fn pure_costs_include_idle_gaps_wait_inn_rollback_and_checked_overflow() {
        let sleep = hexwork_core::services::sleep_ticks();
        let events = vec![
            SessionEvent::action(2, 3, CommittedAction::Wait),
            SessionEvent::action(
                5,
                5 + sleep,
                CommittedAction::UseService {
                    service: Service::Inn,
                },
            ),
            resurrect(5 + sleep, 0, 0, 1),
            SessionEvent::action(1, 2, CommittedAction::Wait),
        ];
        assert_eq!(
            timeline_costs(&events, Mode::Story, 4).unwrap(),
            TimelineCosts {
                full_steps: 9,
                unattended_steps: sleep,
                total_elapsed_ticks: 9 + sleep,
            }
        );
        assert!(timeline_costs(
            &[SessionEvent::action(
                u64::MAX,
                u64::MAX,
                CommittedAction::Wait
            )],
            Mode::Classic,
            u64::MAX,
        )
        .is_err());
        assert!(timeline_costs(
            &[SessionEvent::action(
                0,
                0,
                CommittedAction::UseService {
                    service: Service::Tinker
                }
            )],
            Mode::Classic,
            0,
        )
        .is_err());
    }

    #[test]
    fn structural_helper_virtual_append_checks_prefix_timing_ids_and_event_cap() {
        struct Case {
            name: &'static str,
            identity: RunIdentity,
            current: State,
            checkpoint: Option<StoryCheckpoint>,
            base: Vec<SessionEvent>,
            appended: SessionEvent,
            expected: PersistenceError,
        }

        let seed = 28;
        let (mut story_current, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        // A Story checkpoint ends on the transition at its own tick; the
        // transition now spans its authored ten ticks, so the checkpoint tick
        // is ten.
        story_current.tick = 10;
        let story_checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&story_current).unwrap(),
            after_event_count: 1,
        };
        let appended_transition = transition(0);
        validate_run_structural_limits(
            &identity(Mode::Story, seed),
            &story_current.to_image().unwrap(),
            Some(&story_checkpoint),
            &[],
            Some(&appended_transition),
        )
        .unwrap();

        let (classic_current, _) = player_state(seed, RunLocation::Deep(1));
        let mut wrong_identity = identity(Mode::Classic, seed);
        wrong_identity.display_name = " ".into();
        let mut wrong_player_current =
            State::from_image(classic_current.to_image().unwrap()).unwrap();
        let player_id = wrong_player_current
            .to_image()
            .unwrap()
            .actor_identities
            .iter()
            .find(|(_, actor_identity)| actor_identity.role == ActorRole::Player)
            .map(|(id, _)| *id)
            .unwrap();
        let mut wrong_player_image = wrong_player_current.to_image().unwrap();
        wrong_player_image
            .actor_identities
            .get_mut(&player_id)
            .unwrap()
            .display_name = "Bea".into();
        wrong_player_current = State::from_image(wrong_player_image).unwrap();

        let cases = vec![
            Case {
                name: "event cap",
                identity: identity(Mode::Classic, seed),
                current: State::from_image(classic_current.to_image().unwrap()).unwrap(),
                checkpoint: None,
                base: vec![
                    SessionEvent::action(0, 0, CommittedAction::Move(Direction::North));
                    MAX_SESSION_EVENTS
                ],
                appended: SessionEvent::action(0, 0, CommittedAction::Pickup { item_id: 1 }),
                expected: PersistenceError::Limit {
                    context: "run.events".into(),
                    actual: MAX_SESSION_EVENTS + 1,
                    limit: MAX_SESSION_EVENTS,
                },
            },
            Case {
                name: "wrong transfer timing",
                identity: identity(Mode::Classic, seed),
                current: State::from_image(classic_current.to_image().unwrap()).unwrap(),
                checkpoint: None,
                base: Vec::new(),
                appended: SessionEvent::action(0, 1, CommittedAction::Pickup { item_id: 1 }),
                expected: PersistenceError::Invalid {
                    context: "run.events[0] action span is 1, expected 10".into(),
                },
            },
            Case {
                name: "checkpoint prefix",
                identity: identity(Mode::Story, seed),
                current: State::from_image(story_current.to_image().unwrap()).unwrap(),
                checkpoint: Some(story_checkpoint.clone()),
                base: vec![transition(0)],
                appended: resurrect(10, 10, 2, story_checkpoint.state.hash),
                expected: PersistenceError::Invalid {
                    context:
                        "run.events[1].checkpoint_after_event_count 2 is outside retained prefix 1"
                            .into(),
                },
            },
            Case {
                name: "identity metadata",
                identity: wrong_identity,
                current: State::from_image(classic_current.to_image().unwrap()).unwrap(),
                checkpoint: None,
                base: Vec::new(),
                appended: SessionEvent::action(0, 0, CommittedAction::Drop { item_id: 1 }),
                expected: PersistenceError::Invalid {
                    context: "run.identity.display_name is blank".into(),
                },
            },
            Case {
                name: "Player metadata",
                identity: identity(Mode::Classic, seed),
                current: wrong_player_current,
                checkpoint: None,
                base: Vec::new(),
                appended: SessionEvent::action(
                    0,
                    0,
                    CommittedAction::InspectBoard {
                        town: "milthollow".into(),
                    },
                ),
                expected: PersistenceError::Invalid {
                    context: "run.current Player display name differs from run identity".into(),
                },
            },
        ];

        for case in cases {
            let image = case.current.to_image().unwrap();
            let helper_error = validate_run_structural_limits(
                &case.identity,
                &image,
                case.checkpoint.as_ref(),
                &case.base,
                Some(&case.appended),
            )
            .unwrap_err();
            let mut materialized = case.base;
            materialized.push(case.appended);
            let encode_error = encode_run(
                &case.identity,
                &case.current,
                case.checkpoint.as_ref(),
                &materialized,
            )
            .unwrap_err();
            assert_eq!(helper_error, case.expected, "{} helper", case.name);
            assert_eq!(encode_error, case.expected, "{} encoder", case.name);
        }
    }

    #[test]
    fn transfer_structural_preflight_matches_encoder_checks_at_pack_boundary() {
        let seed = 29;
        let id = identity(Mode::Classic, seed);
        let (state, player) = player_state(seed, RunLocation::Deep(1));
        let mut image = state.to_image().unwrap();
        let mut pack = Pack::new();
        for offset in 0..MAX_PACK_ITEMS as u64 {
            pack.add(test_item(10_000 + offset));
        }
        image.packs.insert(player, pack);
        image.next_id = 10_000 + MAX_PACK_ITEMS as u64;
        let state = State::from_image(image.clone()).unwrap();
        validate_run_structural_limits(&id, &image, None, &[], None).unwrap();
        encode_run(&id, &state, None, &[]).unwrap();

        image
            .packs
            .get_mut(&player)
            .unwrap()
            .add(test_item(10_000 + MAX_PACK_ITEMS as u64));
        image.next_id += 1;
        let state = State::from_image(image.clone()).unwrap();
        let helper_error =
            validate_run_structural_limits(&id, &image, None, &[], None).unwrap_err();
        let encode_error = encode_run(&id, &state, None, &[]).unwrap_err();
        assert_eq!(helper_error, encode_error);
        assert!(matches!(
            helper_error,
            PersistenceError::Limit { context, actual, limit }
                if context == "pack.items"
                    && actual == MAX_PACK_ITEMS + 1
                    && limit == MAX_PACK_ITEMS
        ));
    }

    #[test]
    fn story_aggregate_pack_items_at_cap_and_plus_one_match_encoder() {
        fn image_with_pack_items(
            seed: u64,
            location: RunLocation,
            first_item_id: u64,
        ) -> StateImage {
            let (state, _) = player_state(seed, location);
            let mut image = state.to_image().unwrap();
            let mut next_actor_id = image.next_id;
            let mut next_item_id = first_item_id;
            for pack_index in 0..16 {
                let actor_id = next_actor_id;
                next_actor_id += 1;
                image
                    .actors
                    .push(Actor::new(actor_id, hexwork_core::spatial::FixedVec3::from_cell_feet([pack_index, 0, 0]).unwrap()));
                image.actor_identities.insert(
                    actor_id,
                    ActorIdentity::new("porter", format!("Porter {pack_index}"), ActorRole::Npc),
                );
                image.wallets.insert(actor_id, Wallet::default());
                let mut pack = Pack::new();
                for _ in 0..MAX_PACK_ITEMS {
                    pack.add(test_item(next_item_id));
                    next_item_id += 1;
                }
                image.packs.insert(actor_id, pack);
            }
            image.next_id = next_actor_id.max(next_item_id);
            State::from_image(image.clone()).unwrap();
            image
        }

        let seed = 30;
        let identity = identity(Mode::Story, seed);
        let mut current_image = image_with_pack_items(seed, RunLocation::Deep(1), 10_000);
        let checkpoint_image = image_with_pack_items(
            seed,
            RunLocation::Town("milthollow".into()),
            10_000 + (16 * MAX_PACK_ITEMS) as u64,
        );
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_image(checkpoint_image).unwrap(),
            after_event_count: 0,
        };
        let current = State::from_image(current_image.clone()).unwrap();
        let mut budget = EncodeBudget::default();
        budget.check_identity(&identity, "run.identity").unwrap();
        budget.check_image(&current_image).unwrap();
        budget.check_image(&checkpoint.state.image).unwrap();
        assert_eq!(budget.pack_items, MAX_TOTAL_PACK_ITEMS);
        validate_run_structural_limits(&identity, &current_image, Some(&checkpoint), &[], None)
            .unwrap();
        encode_run(&identity, &current, Some(&checkpoint), &[]).unwrap();

        let extra_owner = current_image.next_id;
        current_image
            .actors
            .push(Actor::new(extra_owner, hexwork_core::spatial::FixedVec3::from_cell_feet([20, 0, 0]).unwrap()));
        current_image.actor_identities.insert(
            extra_owner,
            ActorIdentity::new("porter", "Extra Porter", ActorRole::Npc),
        );
        current_image.wallets.insert(extra_owner, Wallet::default());
        let extra_item_id = extra_owner + 1;
        let mut extra_pack = Pack::new();
        extra_pack.add(test_item(extra_item_id));
        current_image.packs.insert(extra_owner, extra_pack);
        current_image.next_id = extra_item_id + 1;
        let current = State::from_image(current_image.clone()).unwrap();
        let helper_error =
            validate_run_structural_limits(&identity, &current_image, Some(&checkpoint), &[], None)
                .unwrap_err();
        let encode_error = encode_run(&identity, &current, Some(&checkpoint), &[]).unwrap_err();
        assert_eq!(helper_error, encode_error);
        assert!(matches!(
            helper_error,
            PersistenceError::Limit { context, actual, limit }
                if context == "pack items"
                    && actual == MAX_TOTAL_PACK_ITEMS + 1
                    && limit == MAX_TOTAL_PACK_ITEMS
        ));
    }

    #[test]
    fn v4_initial_lineage_and_dual_images_share_all_budgets() {
        let seed = 29;
        let (town, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&town).unwrap(),
            after_event_count: 0,
        };
        let (deep, _) = player_state(seed, RunLocation::Deep(1));
        let mut encode_budget = EncodeBudget {
            chunks: MAX_TOTAL_CHUNKS,
            ..EncodeBudget::default()
        };
        assert!(checked_total(
            &mut encode_budget.chunks,
            1,
            MAX_TOTAL_CHUNKS,
            "total chunks"
        )
        .is_err());
        let mut decode_budget = DecodeBudget {
            chunks: MAX_TOTAL_CHUNKS,
            ..DecodeBudget::default()
        };
        assert!(decode_budget.add(BUDGET_CHUNKS, 1).is_err());
        let bytes =
            encode_run(&identity(Mode::Story, seed), &deep, Some(&checkpoint), &[]).unwrap();
        assert!(decode_run(&bytes, "pack").is_ok());
    }

    #[test]
    fn max_valid_vault_projection_fits_run_and_replay_with_64_plus_64_images() {
        fn populate_64_chunks(state: &mut State) {
            for z in 0..4 {
                for y in 0..4 {
                    for x in 0..4 {
                        state.grid.set_world(
                            x * 48,
                            y * 48,
                            z * 48,
                            Voxel {
                                material: 4,
                                ..Voxel::default()
                            },
                        );
                    }
                }
            }
        }
        fn sized_id(prefix: &str, index: usize) -> String {
            let head = format!("{prefix}:{index:04}:");
            format!("{head}{}", "x".repeat(KNOWLEDGE_ID_MAX_BYTES - head.len()))
        }
        fn ids(prefix: &str, count: usize) -> Vec<String> {
            (0..count).map(|index| sized_id(prefix, index)).collect()
        }
        fn sized_tag(lot: usize, tag: usize) -> String {
            let head = format!("tag:{lot:03}:{tag:02}:");
            format!("{head}{}", "t".repeat(MAX_LOT_TAG_BYTES - head.len()))
        }

        let seed = 30;
        let mut identity = identity(Mode::Story, seed);
        identity.initial_lineage.mapped_regions = ids("map", MAX_MAPPED_REGIONS);
        identity.initial_lineage.recipes = ids("recipe", MAX_RECIPES);
        identity.initial_lineage.translated_hymns = ids("hymn", MAX_TRANSLATIONS);
        identity.initial_lineage.identified_item_templates = ids("item", MAX_IDENTIFIED_TEMPLATES);
        identity.initial_lineage.ghosts = (1..=MAX_GHOSTS)
            .map(|sequence| GhostMemory {
                origin_run_id: RunId::new(sequence as u64).unwrap(),
                item_template_id: sized_id("ghost", sequence),
                warning: "w".repeat(MAX_GHOST_WARNING_BYTES),
            })
            .collect();
        identity.initial_lineage.auction_lots = (1..=MAX_AUCTION_LOTS)
            .map(|item_id| {
                let origin_run_id = RunId::new(item_id as u64).unwrap();
                AuctionLot {
                    lot_id: LotId::new(origin_run_id, item_id as u64).unwrap(),
                    origin_run_id,
                    origin_item_id: item_id as u64,
                    template_id: sized_id("lot", item_id),
                    material: 2,
                    identified: item_id % 2 == 0,
                    cursed: false,
                    blessed: false,
                    condition: Condition::Sound,
                    silvered: false,
                    illegal: false,
                    tags: (0..MAX_LOT_TAGS)
                        .map(|tag| sized_tag(item_id, tag))
                        .collect(),
                    fuel: 0,
                    wet_ticks: 0,
                    base_price_cu: 10,
                    buyback_price_cu: 11,
                }
            })
            .collect();
        validate_initial_lineage(&identity.initial_lineage, true).unwrap();
        assert_eq!(
            identity.initial_lineage.mapped_regions.len(),
            MAX_MAPPED_REGIONS
        );
        assert_eq!(identity.initial_lineage.recipes.len(), MAX_RECIPES);
        assert_eq!(
            identity.initial_lineage.translated_hymns.len(),
            MAX_TRANSLATIONS
        );
        assert_eq!(
            identity.initial_lineage.identified_item_templates.len(),
            MAX_IDENTIFIED_TEMPLATES
        );
        assert_eq!(identity.initial_lineage.ghosts.len(), MAX_GHOSTS);
        assert_eq!(
            identity.initial_lineage.auction_lots.len(),
            MAX_AUCTION_LOTS
        );
        assert!(identity
            .initial_lineage
            .ghosts
            .iter()
            .all(
                |ghost| ghost.item_template_id.len() == KNOWLEDGE_ID_MAX_BYTES
                    && ghost.warning.len() == MAX_GHOST_WARNING_BYTES
            ));
        assert!(identity.initial_lineage.auction_lots.iter().all(|lot| {
            lot.template_id.len() == KNOWLEDGE_ID_MAX_BYTES
                && lot.tags.len() == MAX_LOT_TAGS
                && lot.tags.iter().all(|tag| tag.len() == MAX_LOT_TAG_BYTES)
        }));
        let expected_initial_rows = MAX_MAPPED_REGIONS
            + MAX_RECIPES
            + MAX_TRANSLATIONS
            + MAX_IDENTIFIED_TEMPLATES
            + MAX_GHOSTS
            + MAX_AUCTION_LOTS;
        let expected_initial_strings = (MAX_MAPPED_REGIONS
            + MAX_RECIPES
            + MAX_TRANSLATIONS
            + MAX_IDENTIFIED_TEMPLATES
            + MAX_GHOSTS
            + MAX_AUCTION_LOTS)
            * KNOWLEDGE_ID_MAX_BYTES
            + MAX_GHOSTS * MAX_GHOST_WARNING_BYTES
            + MAX_INITIAL_TAGS * MAX_LOT_TAG_BYTES;
        assert_eq!(expected_initial_rows, 4_736);
        assert_eq!(expected_initial_strings, 1_044_480);

        let (mut town, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        populate_64_chunks(&mut town);
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&town).unwrap(),
            after_event_count: 0,
        };
        drop(town);
        let (mut deep, _) = player_state(seed, RunLocation::Deep(1));
        populate_64_chunks(&mut deep);
        let expected_tick = deep.tick;
        let expected_hash = deep.try_hash_state().unwrap();

        let current_image = deep.to_image().unwrap();
        let mut current_budget = EncodeBudget::default();
        current_budget.check_image(&current_image).unwrap();
        let mut checkpoint_budget = EncodeBudget::default();
        checkpoint_budget
            .check_image(&checkpoint.state.image)
            .unwrap();

        let mut budget = EncodeBudget::default();
        budget.check_identity(&identity, "run.identity").unwrap();
        let strings_after_identity = budget.strings;
        budget.check_image(&current_image).unwrap();
        budget.check_image(&checkpoint.state.image).unwrap();
        drop(current_image);
        assert_eq!(budget.initial_rows, expected_initial_rows);
        assert_eq!(budget.initial_tags, MAX_INITIAL_TAGS);
        assert_eq!(budget.initial_strings, expected_initial_strings);
        assert_eq!(strings_after_identity, expected_initial_strings + 7);
        assert_eq!(budget.chunks, MAX_TOTAL_CHUNKS);
        assert_eq!(
            budget.state_rows,
            expected_initial_rows + current_budget.state_rows + checkpoint_budget.state_rows
        );
        assert_eq!(
            budget.nested_rows,
            expected_initial_rows
                + MAX_INITIAL_TAGS
                + current_budget.nested_rows
                + checkpoint_budget.nested_rows
        );
        assert_eq!(
            budget.strings,
            strings_after_identity + current_budget.strings + checkpoint_budget.strings
        );
        assert_eq!(
            budget.gas_flows,
            current_budget.gas_flows + checkpoint_budget.gas_flows
        );
        assert_eq!(
            budget.mule_waypoints,
            current_budget.mule_waypoints + checkpoint_budget.mule_waypoints
        );
        assert_eq!(
            budget.town_ledger,
            current_budget.town_ledger + checkpoint_budget.town_ledger
        );
        assert_eq!(
            budget.pack_items,
            current_budget.pack_items + checkpoint_budget.pack_items
        );
        assert_eq!(
            budget.hymn_cards,
            current_budget.hymn_cards + checkpoint_budget.hymn_cards
        );
        assert_eq!(
            budget.hymn_miscasts,
            current_budget.hymn_miscasts + checkpoint_budget.hymn_miscasts
        );
        assert_eq!(
            budget.ledger_debts,
            current_budget.ledger_debts + checkpoint_budget.ledger_debts
        );
        assert_eq!(
            budget.shop_rows,
            current_budget.shop_rows + checkpoint_budget.shop_rows
        );
        assert_eq!(
            budget.item_tags,
            current_budget.item_tags + checkpoint_budget.item_tags
        );
        assert_eq!(
            budget.knowledge_rows,
            current_budget.knowledge_rows + checkpoint_budget.knowledge_rows
        );
        assert_eq!(
            budget.knowledge_strings,
            current_budget.knowledge_strings + checkpoint_budget.knowledge_strings
        );
        assert_eq!(budget.nested_rows, 14_168_704);
        assert_eq!(budget.state_rows, 4_872);
        assert_eq!(budget.strings, 1_044_515);
        assert!(budget.strings <= MAX_TOTAL_STRING_BYTES);
        assert!(budget.state_rows <= MAX_TOTAL_STATE_ROWS);
        assert!(budget.nested_rows <= MAX_TOTAL_NESTED_ROWS);

        let bytes = encode_run(&identity, &deep, Some(&checkpoint), &[]).unwrap();
        assert!(bytes.len() <= MAX_COMPRESSED_BYTES);
        let raw = decompress_bounded(&bytes).unwrap();
        assert!(raw.len() <= MAX_DECOMPRESSED_BYTES);
        eprintln!(
            "max projection 64+64: strings={} rows={} tags={} nested={} compressed={} raw={}",
            budget.strings,
            budget.initial_rows,
            budget.initial_tags,
            budget.nested_rows,
            bytes.len(),
            raw.len()
        );
        drop(raw);
        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.identity.initial_lineage, identity.initial_lineage);
        assert_eq!(decoded.current.grid.iter_loaded().count(), 64);
        assert_eq!(
            decoded
                .story_checkpoint
                .as_ref()
                .unwrap()
                .state
                .image
                .chunks
                .len(),
            64
        );
        drop(decoded);
        drop(bytes);
        drop(deep);

        let replay = ProductionReplay {
            identity,
            events: Vec::new(),
            expected_final_tick: expected_tick,
            expected_final_hash: expected_hash,
        };
        let bytes = encode_replay(&replay).unwrap();
        assert!(bytes.len() <= MAX_COMPRESSED_BYTES);
        let raw = decompress_bounded(&bytes).unwrap();
        assert!(raw.len() <= MAX_DECOMPRESSED_BYTES);
        drop(raw);
        let decoded = decode_replay(&bytes, "pack").unwrap();
        assert_eq!(
            decoded.identity.initial_lineage,
            replay.identity.initial_lineage
        );
    }

    #[test]
    fn actual_shared_string_budget_rejects_whole_run_encode_and_decode() {
        const LINKS_PER_IMAGE: usize = 2_049;

        fn state_with_individually_valid_strings(seed: u64, location: RunLocation) -> State {
            let (mut state, _) = player_state(seed, location);
            let detail = "d".repeat(MAX_STRING_BYTES);
            for tick in 0..LINKS_PER_IMAGE as u64 {
                state
                    .causes
                    .push(CauseLink::new(tick, "gravity", detail.clone()));
            }
            state
        }

        let seed = 32;
        let current = state_with_individually_valid_strings(seed, RunLocation::Deep(1));
        let checkpoint_state =
            state_with_individually_valid_strings(seed, RunLocation::Town("milthollow".into()));
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&checkpoint_state).unwrap(),
            after_event_count: 0,
        };
        drop(checkpoint_state);

        let mut current_only = EncodeBudget::default();
        current_only
            .check_image(&current.to_image().unwrap())
            .unwrap();
        let mut checkpoint_only = EncodeBudget::default();
        checkpoint_only
            .check_image(&checkpoint.state.image)
            .unwrap();
        assert!(current_only.strings < MAX_TOTAL_STRING_BYTES);
        assert!(checkpoint_only.strings < MAX_TOTAL_STRING_BYTES);
        assert!(current_only.strings + checkpoint_only.strings > MAX_TOTAL_STRING_BYTES);

        let identity = identity(Mode::Story, seed);
        let current_image = current.to_image().unwrap();
        let helper_error =
            validate_run_structural_limits(&identity, &current_image, Some(&checkpoint), &[], None)
                .unwrap_err();
        let encode_error = encode_run(&identity, &current, Some(&checkpoint), &[]).unwrap_err();
        assert_eq!(helper_error, encode_error);
        assert!(matches!(
            encode_error,
            PersistenceError::Limit { context, actual, limit }
                if context == "total strings"
                    && actual > MAX_TOTAL_STRING_BYTES
                    && limit == MAX_TOTAL_STRING_BYTES
        ));

        let wire = run_wire(&identity, &current, Some(&checkpoint), &[]);
        let raw = bincode::encode_to_vec(
            wire,
            bincode::config::standard().with_limit::<MAX_DECOMPRESSED_BYTES>(),
        )
        .unwrap();
        let compressed = zstd::stream::encode_all(raw.as_slice(), ZSTD_LEVEL).unwrap();
        drop(raw);
        let decode_error = decode_run(&compressed, "pack").unwrap_err();
        assert!(matches!(
            decode_error,
            PersistenceError::Decode(ref message)
                if message.contains("total strings aggregate")
                    && message.contains("exceeds limit")
        ));
    }

    #[test]
    fn actual_54_plus_54_images_roundtrip_within_envelope_caps() {
        fn populate_54_chunks(state: &mut State) {
            for z in -3..3 {
                for y in -1..=1 {
                    for x in -1..=1 {
                        state.grid.set_world(
                            x * 48,
                            y * 48,
                            z * 48,
                            Voxel {
                                material: 4,
                                ..Voxel::default()
                            },
                        );
                    }
                }
            }
        }

        let seed = 31;
        let (mut town, _) = player_state(seed, RunLocation::Town("milthollow".into()));
        populate_54_chunks(&mut town);
        let checkpoint = StoryCheckpoint {
            state: HashedState::from_state(&town).unwrap(),
            after_event_count: 0,
        };
        assert_eq!(checkpoint.state.image.chunks.len(), 54);
        let expected_checkpoint_hash = checkpoint.state.hash;
        drop(town);

        let (mut deep, _) = player_state(seed, RunLocation::Deep(1));
        populate_54_chunks(&mut deep);
        let expected_current_hash = deep.try_hash_state().unwrap();
        assert_eq!(deep.grid.iter_loaded().count(), 54);

        let current_image = deep.to_image().unwrap();
        let mut budget = EncodeBudget::default();
        budget.check_image(&current_image).unwrap();
        budget.check_image(&checkpoint.state.image).unwrap();
        assert_eq!(budget.chunks, 108);
        assert_eq!(budget.nested_rows, 108 * CHUNK_VOL);
        assert!(budget.chunks <= MAX_TOTAL_CHUNKS);
        assert!(budget.state_rows <= MAX_TOTAL_STATE_ROWS);
        assert!(budget.nested_rows <= MAX_TOTAL_NESTED_ROWS);
        drop(current_image);

        let bytes =
            encode_run(&identity(Mode::Story, seed), &deep, Some(&checkpoint), &[]).unwrap();
        assert!(bytes.len() <= MAX_COMPRESSED_BYTES);
        let raw = decompress_bounded(&bytes).unwrap();
        assert!(raw.len() <= MAX_DECOMPRESSED_BYTES);
        eprintln!(
            "54+54 Story envelope: compressed={} raw={}",
            bytes.len(),
            raw.len()
        );
        drop(raw);

        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.current.grid.iter_loaded().count(), 54);
        assert_eq!(
            decoded.current.try_hash_state().unwrap(),
            expected_current_hash
        );
        let decoded_checkpoint = decoded.story_checkpoint.as_ref().unwrap();
        assert_eq!(decoded_checkpoint.state.image.chunks.len(), 54);
        assert_eq!(decoded_checkpoint.state.hash, expected_checkpoint_hash);
        assert_eq!(decoded_checkpoint.after_event_count, 0);
        assert!(matches!(
            decoded_checkpoint.state.image.run_location,
            RunLocation::Town(ref town) if town == "milthollow"
        ));
    }

    #[test]
    fn bounded_prefixes_and_aggregate_helpers_reject_without_large_allocations() {
        fn decode_prefix<T: Decode<DecodeBudget>>(length: u64) -> Result<T, DecodeError> {
            let bytes = bincode::encode_to_vec(length, bincode::config::standard()).unwrap();
            bincode::decode_from_slice_with_context(
                &bytes,
                bincode::config::standard(),
                DecodeBudget::default(),
            )
            .map(|(value, _)| value)
        }
        assert!(decode_prefix::<BoundedVec<u8, 3, BUDGET_NONE>>(4).is_err());
        assert!(decode_prefix::<BoundedString>((MAX_STRING_BYTES + 1) as u64).is_err());
        let mut total = MAX_TOTAL_STATE_ROWS;
        assert!(DecodeBudget::add_counter(&mut total, 1, MAX_TOTAL_STATE_ROWS, "rows").is_err());
        let mut total = MAX_TOTAL_STATE_ROWS;
        assert!(checked_total(&mut total, 1, MAX_TOTAL_STATE_ROWS, "rows").is_err());
    }

    #[test]
    fn encode_rejects_oversized_nested_containers_before_wire_construction() {
        let mut image = State::new(1).to_image().unwrap();
        image.items.insert(
            1,
            Instance {
                id: 1,
                template: "item".into(),
                material: 0,
                identified: false,
                cursed: false,
                blessed: false,
                condition: Condition::Sound,
                silvered: false,
                illegal: false,
                tags: vec!["tag".into(); MAX_ITEM_TAGS + 1],
                fuel: 0,
                wet_ticks: 0,
            },
        );
        assert!(
            matches!(image_to_wire(image), Err(PersistenceError::Limit { context, .. }) if context == "item.tags")
        );
    }

    #[test]
    fn crafted_reputation_wire_rejects_duplicate_and_reordered_arrays() {
        let mut state = State::new(37);
        let player = state.spawn_actor_with_identity(
            hexwork_core::spatial::FixedVec3::from_cell_feet([0; 3]).unwrap(),
            ActorIdentity::new("delver", "Ada", ActorRole::Player),
        );
        state.reputations.insert(player, Reputation::new());
        let mut wire = image_to_wire(state.to_image().unwrap()).unwrap();
        wire.reputations[0].1.factions.swap(0, 1);
        assert!(
            matches!(wire_to_image(wire), Err(PersistenceError::Invalid { context }) if context.contains("factions"))
        );
    }

    #[test]
    fn fixed_wire_body_bounds_and_staged_moves() {
        let (state, player) = player_state(1, RunLocation::Town("milthollow".into()));
        let mut image = state.to_image().unwrap();
        image.spill.desired_moves.insert(player, hexwork_core::locomotion::MovementSample::player(Some(hexwork_core::heading::Heading::WEST)));
        image.spill.desired_moves.get_mut(&player).unwrap().stop_at=Some(hexwork_core::spatial::FixedVec3::new([-123,456,0]).unwrap());
        image.actors[0].heading=Some(hexwork_core::heading::Heading::new(17).unwrap());
        image.actors[0].movement_remainder=hexwork_core::heading::MovementRemainder::new([123,-456]).unwrap();
        image.spill.crossed_cells.entry(player).or_default().insert((-1,2,0));
        let back = wire_to_image(image_to_wire(image.clone()).unwrap()).unwrap();
        assert_eq!(back.spill.desired_moves, image.spill.desired_moves);
        assert_eq!(back.spill.crossed_cells,image.spill.crossed_cells);
        assert_eq!(back.actors[0].heading,image.actors[0].heading);
        assert_eq!(back.actors[0].movement_remainder,image.actors[0].movement_remainder);
        State::from_image(back).unwrap();
        let mut wire = image_to_wire(image.clone()).unwrap();
        wire.actors[0].position_mm[0] = hexwork_core::spatial::MAX_POSITION_MM;
        let back = wire_to_image(wire).unwrap();
        assert!(State::from_image(back).is_err());
        let mut wire = image_to_wire(image.clone()).unwrap();
        wire.actors[0].body_mm[0] = 0;
        assert!(wire_to_image(wire).is_err());
        let mut wire = image_to_wire(image.clone()).unwrap();
        wire.actors[0].movement_remainder=[i64::MAX,0];
        assert!(wire_to_image(wire).is_err());
        let mut wire = image_to_wire(image).unwrap();
        wire.spill.desired_moves = vec![(player, Some(4096), 3000, None)].into_iter().collect();
        assert!(wire_to_image(wire).is_err());
    }

    #[test]
    fn locomotion_serialized_resume_matches_live_state() {
        use hexwork_core::{heading::Heading,voxel::{Voxel,material}};
        let (mut state,id)=player_state(19,RunLocation::Town("milthollow".into()));
        for x in -2..=3 {for y in -2..=3 {state.grid.set_world(x,y,-1,Voxel {material:material::STONE,..Voxel::default()});}}
        let heading=Some(Heading::new(512).unwrap());
        state.admit_player_movement(id,heading).unwrap();state.step();
        state.admit_player_movement(id,heading).unwrap();
        let wire=image_to_wire(state.to_image().unwrap()).unwrap();
        let raw=bincode::encode_to_vec(wire,bincode::config::standard()).unwrap();
        let wire: StateWire=decode_raw_bounded(&raw).unwrap();
        let mut restored=State::from_image(wire_to_image(wire).unwrap()).unwrap();
        state.step();restored.step();
        assert_eq!(state.actors[0].position,restored.actors[0].position);
        assert_eq!(state.actors[0].movement_remainder,restored.actors[0].movement_remainder);
        assert_eq!(state.hash_state(),restored.hash_state());
    }

    #[test]
    fn fixed_staged_moves_aggregate_budget() {
        let mut image = State::new(1).to_image().unwrap();
        let mut baseline = EncodeBudget::default();
        baseline.check_image(&image).unwrap();
        image.spill.desired_moves.insert(1, hexwork_core::locomotion::MovementSample::player(Some(hexwork_core::heading::Heading::EAST)));
        let mut exact = EncodeBudget { state_rows: MAX_TOTAL_STATE_ROWS - baseline.state_rows - 1, ..EncodeBudget::default() };
        exact.check_image(&image).unwrap();
        assert_eq!(exact.state_rows, MAX_TOTAL_STATE_ROWS);
        let mut exhausted = EncodeBudget { state_rows: MAX_TOTAL_STATE_ROWS - baseline.state_rows, ..EncodeBudget::default() };
        assert!(exhausted.check_image(&image).is_err());
        let rows: BoundedVec<(u64, Option<u16>, u32, Option<[i64;3]>), MAX_ROWS, BUDGET_STATE_ROWS> = vec![(1, Some(1024), 3000, None)].into_iter().collect();
        let raw = bincode::encode_to_vec(rows, bincode::config::standard()).unwrap();
        let decoded = bincode::decode_from_slice_with_context::<DecodeBudget, BoundedVec<(u64, Option<u16>, u32, Option<[i64;3]>), MAX_ROWS, BUDGET_STATE_ROWS>, _>(
            &raw, bincode::config::standard(), DecodeBudget { state_rows: MAX_TOTAL_STATE_ROWS, ..DecodeBudget::default() });
        assert!(decoded.is_err());
    }

    #[test]
    fn f32_bits_roundtrip() {
        let mut image = State::new(1).to_image().unwrap();
        image.fall_odo.insert(1, u64::MAX);
        image.pressure_nodes.push(PressureNode {
            id: 1,
            pressure: f32::from_bits(1),
        });
        let back = wire_to_image(image_to_wire(image).unwrap()).unwrap();
        assert_eq!(back.fall_odo[&1], u64::MAX);
        assert_eq!(back.pressure_nodes[0].pressure.to_bits(), 1);
    }

    #[test]
    fn descending_chunks_are_not_sorted() {
        let mut image = State::new(1).to_image().unwrap();
        let voxels = vec![Voxel::default(); CHUNK_VOL];
        image.chunks = vec![
            ChunkImage {
                coord: (1, 0, 0),
                voxels: voxels.clone(),
            },
            ChunkImage {
                coord: (0, 0, 0),
                voxels,
            },
        ];
        let back = wire_to_image(image_to_wire(image).unwrap()).unwrap();
        assert_eq!(back.chunks[0].coord, (1, 0, 0));
        assert!(State::from_image(back).is_err());
    }

    #[test]
    fn descending_laws_are_not_sorted() {
        let mut image = State::new(1).to_image().unwrap();
        image.laws = vec![
            LawImage {
                rule: "waste nothing".into(),
                first_seen_tick: 1,
            },
            LawImage {
                rule: "gravity is fair".into(),
                first_seen_tick: 2,
            },
        ];
        let back = wire_to_image(image_to_wire(image).unwrap()).unwrap();
        assert_eq!(back.laws[0].rule, "waste nothing");
        assert!(State::from_image(back).is_err());
    }

    #[test]
    fn run_metadata_content_and_versions_are_gated() {
        let (state, _) = player_state(41, RunLocation::Deep(1));
        let id = identity(Mode::Classic, 41);
        let bytes = encode_run(&id, &state, None, &[]).unwrap();
        assert!(matches!(
            decode_run(&bytes, "other"),
            Err(PersistenceError::ContentHashMismatch { .. })
        ));
        let mut wire = run_wire(&id, &state, None, &[]);
        wire.header.format_version += 1;
        assert!(matches!(
            decode_run(&raw(wire), "pack"),
            Err(PersistenceError::NewerVersion { .. })
        ));
        let mut wire = run_wire(&id, &state, None, &[]);
        wire.header.canonical_hash_version += 1;
        assert!(matches!(
            decode_run(&raw(wire), "pack"),
            Err(PersistenceError::CanonicalHashVersion { .. })
        ));
    }

    #[test]
    fn all_actions_roundtrip_and_same_tick_order() {
        let actions = vec![
            CommittedAction::Move(Direction::North),
            CommittedAction::Wait,
            CommittedAction::InspectBoard {
                town: "milthollow".into(),
            },
            CommittedAction::AcceptContract { id: "rat-1".into() },
            CommittedAction::UseTransition { transition_id: 7 },
            CommittedAction::ResolveAttack { target_actor: 8 },
            CommittedAction::TurnIn {
                contract_id: "rat-1".into(),
            },
            CommittedAction::UseService {
                service: Service::Inn,
            },
        ];
        for action in actions {
            assert_eq!(
                wire_to_action(action_to_wire(&action).unwrap(), "action").unwrap(),
                action
            );
        }
    }

    /// Two sequential committed transfers round-trip through a run save and a
    /// replay with their journal order preserved. Each Pickup/Drop now spans its
    /// authored ten ticks, so the events run 0..10..20.
    #[test]
    fn transfer_run_replay_roundtrip_preserves_event_order() {
        let seed = 42;
        let id = identity(Mode::Classic, seed);
        let (mut state, _) = player_state(seed, RunLocation::Deep(1));
        state.tick = 20;
        let events = vec![
            SessionEvent::action(0, 10, CommittedAction::Pickup { item_id: 17 }),
            SessionEvent::action(10, 20, CommittedAction::Drop { item_id: 17 }),
        ];
        let bytes = encode_run(&id, &state, None, &events).unwrap();
        let decoded = decode_run(&bytes, "pack").unwrap();
        assert_eq!(decoded.events, events);

        let replay = ProductionReplay {
            identity: id,
            events,
            expected_final_tick: 20,
            expected_final_hash: state.try_hash_state().unwrap(),
        };
        assert_eq!(
            decode_replay(&encode_replay(&replay).unwrap(), "pack").unwrap(),
            replay
        );
    }

    #[test]
    fn structural_admission_does_not_guarantee_encoded_byte_limits() {
        let seed = 44;
        let id = identity(Mode::Classic, seed);
        let (mut state, _) = player_state(seed, RunLocation::Deep(1));
        state.tick = 10;
        let image = state.to_image().unwrap();
        let events = [SessionEvent::action(0, 10, CommittedAction::Pickup { item_id: 1 })];
        validate_run_structural_limits(&id, &image, None, &events, None).unwrap();
        let wire = run_wire(&id, &state, None, &events);
        assert!(matches!(
            encode_bounded_with_limits(wire, 1, MAX_COMPRESSED_BYTES),
            Err(PersistenceError::Limit { context, limit: 1, .. })
                if context == "encoded envelope"
        ));
        let wire = run_wire(&id, &state, None, &events);
        assert!(matches!(
            encode_bounded_with_limits(wire, MAX_DECOMPRESSED_BYTES, 1),
            Err(PersistenceError::InputTooLarge { limit: 1, .. })
        ));
    }

    #[test]
    fn corrupt_truncated_trailing_and_bomb_err() {
        for bytes in [b"random".as_slice(), b"".as_slice()] {
            assert!(decode_run(bytes, "pack").is_err());
        }
        let (state, _) = player_state(43, RunLocation::Deep(1));
        let id = identity(Mode::Classic, 43);
        let bytes = encode_run(&id, &state, None, &[]).unwrap();
        assert!(decode_run(&bytes[..bytes.len() / 2], "pack").is_err());
        let mut encoded = bincode::encode_to_vec(
            run_wire(&id, &state, None, &[]),
            bincode::config::standard(),
        )
        .unwrap();
        encoded.push(1);
        assert!(matches!(
            decode_run(
                &zstd::stream::encode_all(encoded.as_slice(), 1).unwrap(),
                "pack"
            ),
            Err(PersistenceError::TrailingBytes { .. })
        ));
        let zeroes = std::io::repeat(0).take((MAX_DECOMPRESSED_BYTES as u64) + 1);
        let bomb = zstd::stream::encode_all(zeroes, 1).unwrap();
        assert!(matches!(
            decode_run(&bomb, "pack"),
            Err(PersistenceError::DecompressedTooLarge { .. })
        ));
    }
}