betteroffice-xlsx 0.3.0

use std::borrow::Cow;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::Arc;

use crate::sheet_json::{decode_charts, decode_hyperlinks};
use sha2::{Digest, Sha256};
use xlsx_model::{
    AnchorEditAs, AnchorExtent, AnchorPos, Cell, CellFormat, CellRange, CellRef, CellValue,
    ChartAnchor, ChartRef, ColStyle, DateSystem, DefinedName, ErrorValue, FreezePane, Hyperlink,
    MAX_COLS, MAX_ROWS, Sheet, SheetChart, SheetFormat, SheetId, Stylesheet, Table,
    Workbook as WorkbookModel,
};
use xlsx_ops::{CellState, Op};
use yrs::block::{
    BLOCK_GC_REF_NUMBER, BLOCK_ITEM_ANY_REF_NUMBER, BLOCK_ITEM_DELETED_REF_NUMBER,
    BLOCK_ITEM_TYPE_REF_NUMBER, BLOCK_SKIP_REF_NUMBER, ClientID,
};
use yrs::encoding::read::{Error as DecodeError, Read};
use yrs::encoding::write::Write;
use yrs::sync::time::Clock;
use yrs::types::{TYPE_REFS_ARRAY, TYPE_REFS_MAP};
use yrs::undo::{Options as UndoOptions, StackItem, UndoManager};
use yrs::updates::decoder::{Decode, Decoder, DecoderV1};
use yrs::updates::encoder::{Encoder, EncoderV1};
use yrs::{
    Any, Array, ArrayRef, BranchID, Doc, ID, Map, MapPrelim, MapRef, Options, Origin, Out, ReadTxn,
    StateVector, Transact, TransactionMut, Update, WriteTxn,
};

const META: &str = "xlsx";
const CELL_FORMATS: &str = "xlsx:cell-formats";
const SHEET_ORDER: &str = "xlsx:sheet-order";
const SHEETS: &str = "xlsx:sheets";
const MIN_SUPPORTED_SCHEMA_VERSION: i64 = 3;
/// The schema each feature first appeared in. These are frozen: gating a
/// feature on the current version instead would silently reclassify the
/// newest schema as predating it the moment the current version moves on.
const FREEZE_PANE_SCHEMA_VERSION: i64 = 4;
const HYPERLINK_SCHEMA_VERSION: i64 = 5;
const CHARTS_SCHEMA_VERSION: i64 = 6;
const SCHEMA_VERSION: i64 = 6;
const BASE_FINGERPRINT: &str = "baseFingerprint";
const STRUCTURE_GENERATION: &str = "structureGeneration";
const CHARTS: &str = "charts";
const CONTENTS: &str = "contents";
const COL_WIDTHS: &str = "colWidths";
const FREEZE_PANE: &str = "freezePane";
const HYPERLINKS: &str = "hyperlinks";
const MERGES: &str = "merges";
const NAME: &str = "name";
const ROW_HEIGHTS: &str = "rowHeights";
const STYLES: &str = "styles";
const BOOTSTRAP_ORIGIN: &str = "xlsx:bootstrap";
const HYDRATE_ORIGIN: &str = "xlsx:hydrate";
const REMOTE_ORIGIN: &str = "xlsx:remote";
/// Local adoptions no undo manager captures.
const UNTRACKED_ORIGIN: &str = "xlsx:untracked";
const MAX_SAFE_CLIENT_ID: u64 = (1_u64 << 53) - 1;
const MAX_SAFE_CLOCK: u32 = i32::MAX as u32;
const MAX_UPDATE_BLOCKS: usize = 1_000_000;
const MAX_UPDATE_VALUES: usize = 1_000_000;
const MAX_UPDATE_DELETE_RANGES: usize = 1_000_000;
/// upper bound on one encoded cell format. the canonical form of any format a
/// workbook can hold is three orders of magnitude smaller.
const MAX_CELL_FORMAT_BYTES: usize = 64 * 1024;
const UNDO_CAPTURE_TIMEOUT_MS: u64 = 500;
pub(crate) const MAX_STATE_VECTOR_ENTRIES: u32 = 65_536;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum SyncOrigin {
    User,
    Agent,
    Undo,
    Redo,
}

impl SyncOrigin {
    fn as_str(self) -> &'static str {
        match self {
            Self::User => "xlsx:user",
            Self::Agent => "xlsx:agent",
            Self::Undo => "xlsx:undo",
            Self::Redo => "xlsx:redo",
        }
    }
}

#[derive(Debug)]
pub(crate) enum AuthorityError {
    ClientIdConflict(u64),
    InvalidStateVector(String),
    InvalidUpdate(String),
    InvalidState(String),
}

#[derive(Clone)]
struct WorkbookBase {
    bootstrap_client_id: u64,
    date_system: DateSystem,
    defined_names: Vec<DefinedName>,
    fingerprint: String,
    fingerprints: BTreeMap<i64, Vec<String>>,
    freeze_panes: Vec<Option<FreezePane>>,
    formats: Vec<SheetFormat>,
    col_styles: Vec<Vec<ColStyle>>,
    hyperlinks: Vec<Vec<Hyperlink>>,
    charts: Vec<Vec<SheetChart>>,
    hidden_dimensions: Vec<HiddenDimensions>,
    shared_strings: Vec<String>,
    styles: Stylesheet,
    /// table parts; read-only reference data, not shared state.
    tables: Vec<Table>,
}

impl WorkbookBase {
    /// A legacy fingerprint hashes no chart state, so a charted workbook pairs
    /// with one on its other content alone and keeps the charts it parsed. It
    /// is the only reading of a state written before charts were shared, and
    /// refusing it would strand every snapshot an earlier release persisted.
    #[cfg(test)]
    fn from_model(model: &WorkbookModel) -> Result<Self, String> {
        Self::from_model_with_legacy_dimensions(model, &[], None)
    }

    /// `legacy_dimensions` are the row heights and column widths releases
    /// before hidden rows and columns read as zero stored, per sheet. Those
    /// maps are hashed at every schema, so a peer that persisted its state
    /// under one of those releases is only recognisable against them.
    fn from_model_with_legacy_dimensions(
        model: &WorkbookModel,
        legacy_dimensions: &[xlsx_parse::LegacySheetDimensions],
        legacy_styles: Option<&Stylesheet>,
    ) -> Result<Self, String> {
        let (fingerprint, bootstrap_client_id) = fingerprint_model(model)?;
        let mut fingerprints = BTreeMap::new();
        for version in MIN_SUPPORTED_SCHEMA_VERSION..=SCHEMA_VERSION {
            let (version_fingerprint, _) = fingerprint_model_for_schema(model, version)?;
            fingerprints.insert(version, vec![version_fingerprint]);
        }
        if let Some(version_3) = fingerprints.get_mut(&3) {
            let (defined_names_v3, _) = fingerprint_model_with_schema(model, 3, true, true)?;
            if !version_3.contains(&defined_names_v3) {
                version_3.push(defined_names_v3);
            }
        }
        for version in MIN_SUPPORTED_SCHEMA_VERSION..=SCHEMA_VERSION {
            let (without_col_styles, _) =
                fingerprint_model_with_schema(model, version, version >= 4, false)?;
            let accepted = fingerprints.entry(version).or_default();
            if !accepted.contains(&without_col_styles) {
                accepted.push(without_col_styles);
            }
        }
        if let Some(legacy) = model_with_legacy_dimensions(model, legacy_dimensions) {
            for version in MIN_SUPPORTED_SCHEMA_VERSION..SCHEMA_VERSION {
                let (legacy_fingerprint, _) = fingerprint_model_for_schema(&legacy, version)?;
                let (without_col_styles, _) =
                    fingerprint_model_with_schema(&legacy, version, version >= 4, false)?;
                let accepted = fingerprints.entry(version).or_default();
                for fingerprint in [legacy_fingerprint, without_col_styles] {
                    if !accepted.contains(&fingerprint) {
                        accepted.push(fingerprint);
                    }
                }
            }
            if let Some(version_3) = fingerprints.get_mut(&3) {
                let (defined_names_v3, _) = fingerprint_model_with_schema(&legacy, 3, true, true)?;
                if !version_3.contains(&defined_names_v3) {
                    version_3.push(defined_names_v3);
                }
            }
        }
        if let Some(styles) = legacy_styles.filter(|styles| **styles != model.styles) {
            let mut legacy_model = model.clone();
            legacy_model.styles = styles.clone();
            let legacy_base =
                Self::from_model_with_legacy_dimensions(&legacy_model, legacy_dimensions, None)?;
            for (version, legacy_fingerprints) in legacy_base.fingerprints {
                let accepted = fingerprints.entry(version).or_default();
                for fingerprint in legacy_fingerprints {
                    if !accepted.contains(&fingerprint) {
                        accepted.push(fingerprint);
                    }
                }
            }
        }
        if !model.styles.indexed_colors.is_empty() {
            let mut legacy_model = model.clone();
            legacy_model.styles.indexed_colors.clear();
            let legacy_base = Self::from_model_with_legacy_dimensions(
                &legacy_model,
                legacy_dimensions,
                legacy_styles,
            )?;
            for (version, legacy_fingerprints) in legacy_base.fingerprints {
                let accepted = fingerprints.entry(version).or_default();
                for fingerprint in legacy_fingerprints {
                    if !accepted.contains(&fingerprint) {
                        accepted.push(fingerprint);
                    }
                }
            }
        }
        Ok(Self {
            bootstrap_client_id,
            date_system: model.date_system,
            defined_names: model.defined_names.clone(),
            fingerprint,
            fingerprints,
            freeze_panes: model.sheets.iter().map(|sheet| sheet.freeze_pane).collect(),
            formats: model.sheets.iter().map(|sheet| sheet.format).collect(),
            col_styles: model
                .sheets
                .iter()
                .map(|sheet| sheet.col_styles.clone())
                .collect(),
            hyperlinks: model
                .sheets
                .iter()
                .map(|sheet| sheet.hyperlinks.clone())
                .collect(),
            charts: model
                .sheets
                .iter()
                .map(|sheet| sheet.charts.clone())
                .collect(),
            hidden_dimensions: hidden_dimensions(model, legacy_dimensions),
            shared_strings: model.shared_strings.clone(),
            styles: model.styles.clone(),
            tables: model.tables.clone(),
        })
    }

    fn accepts_fingerprint(&self, version: i64, fingerprint: &str) -> bool {
        self.fingerprints
            .get(&version)
            .is_some_and(|accepted| accepted.iter().any(|value| value == fingerprint))
    }

    fn workbook(&self) -> WorkbookModel {
        WorkbookModel {
            sheets: Vec::new(),
            date_system: self.date_system,
            defined_names: self.defined_names.clone(),
            shared_strings: self.shared_strings.clone(),
            styles: self.styles.clone(),
            tables: self.tables.clone(),
        }
    }
}

/// The part of an anchor no move may rewrite: its kind, plus whatever the
/// drawing writer cannot patch. Only the grid markers a save writes whole stay
/// free, so this mirrors `only_grid_position_moved` in the writer.
#[derive(Clone, Debug, PartialEq, Eq)]
enum AnchorShape {
    TwoCell {
        edit_as: AnchorEditAs,
    },
    OneCell {
        extent: AnchorExtent,
    },
    Absolute {
        pos: AnchorPos,
        extent: AnchorExtent,
    },
}

impl AnchorShape {
    fn of(anchor: &ChartAnchor) -> Self {
        match anchor {
            ChartAnchor::TwoCell { edit_as, .. } => Self::TwoCell { edit_as: *edit_as },
            ChartAnchor::OneCell { extent, .. } => Self::OneCell { extent: *extent },
            ChartAnchor::Absolute { pos, extent } => Self::Absolute {
                pos: *pos,
                extent: *extent,
            },
        }
    }
}

/// What the freeze pins about a chart: which drawing anchors which part, what
/// the part reads, and the shape of that anchor. Only where a grid-anchored
/// chart sits is replicated content, so a peer may slide one without changing
/// the structure this describes.
#[derive(Clone, Debug, PartialEq, Eq)]
struct ChartIdentity {
    part: String,
    drawing: String,
    anchor_index: usize,
    refs: Vec<ChartRef>,
    anchor: AnchorShape,
}

impl ChartIdentity {
    fn of(chart: &SheetChart) -> Self {
        Self {
            part: chart.part.clone(),
            drawing: chart.drawing.clone(),
            anchor_index: chart.anchor_index,
            refs: chart.refs.clone(),
            anchor: AnchorShape::of(&chart.anchor),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct WorkbookStructure {
    generation: i64,
    pub(crate) sheet_keys: Vec<String>,
    sheet_names: Vec<String>,
    freeze_panes: Vec<Option<FreezePane>>,
    hyperlinks: Vec<Vec<Hyperlink>>,
    charts: Vec<Vec<ChartIdentity>>,
    merges: Vec<Vec<CellRange>>,
    shared_types: BTreeMap<String, SheetSharedTypes>,
}

impl WorkbookStructure {
    /// Whether two structures describe the same workbook, disregarding the Yrs
    /// branch identities. Replacing a bootstrap rebuilds every shared type, so
    /// those always differ and cannot say whether the structure itself did.
    pub(crate) fn describes_same_workbook(&self, other: &Self) -> bool {
        self.sheet_keys == other.sheet_keys
            && self.sheet_names == other.sheet_names
            && self.freeze_panes == other.freeze_panes
            && self.hyperlinks == other.hyperlinks
            && self.charts == other.charts
            && self.merges == other.merges
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct SheetSharedTypes {
    sheet: BranchID,
    col_widths: BranchID,
    contents: BranchID,
    row_heights: BranchID,
    styles: BranchID,
}

/// What a replica can do with an update offered as its whole state.
pub(crate) enum SnapshotAdoption {
    /// Not a whole document, or this replica already holds more than its own
    /// bootstrap. The update is an ordinary one; merge it.
    NotApplicable,
    /// A whole document this replica cannot take on.
    Incompatible(String),
    Replacement(Box<WorkbookAuthority>),
}

pub(crate) struct StagedUpdate {
    pub(crate) commit_update: Vec<u8>,
    pub(crate) effective: bool,
    pub(crate) model: WorkbookModel,
    pub(crate) pending: bool,
    pub(crate) state_bytes: usize,
    pub(crate) state_vector_entries: usize,
    pub(crate) structure: WorkbookStructure,
    pub(crate) update: Vec<u8>,
}

pub(crate) struct StagedLocalUpdate {
    pub(crate) model: WorkbookModel,
    pub(crate) state_bytes: usize,
    pub(crate) state_vector_entries: usize,
    pub(crate) structure: WorkbookStructure,
    pub(crate) update: Vec<u8>,
    carried: CarriedState,
}

/// What staging changed beside the document, handed to the live authority on adoption.
struct CarriedState {
    base: Arc<WorkbookBase>,
    next_sheet_id: u64,
    sheet_order: Option<SheetOrderEntry>,
}

/// Which history an adopted local update enters.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum LocalHistory {
    /// One collaborative undo item.
    Undo,
    /// The sheet-order entry a standalone undo replays against.
    SheetOrder,
    /// Neither; existing undo and redo entries stay.
    None,
}

pub(crate) struct AuthorityCheckpoint {
    state: Vec<u8>,
    undo_stack: Vec<StackItem<()>>,
    redo_stack: Vec<StackItem<()>>,
}

pub(crate) struct HistoryUpdate {
    pub(crate) model: WorkbookModel,
    pub(crate) structure: WorkbookStructure,
    pub(crate) update: Vec<u8>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct SheetOrderEntry {
    before: Vec<String>,
    after: Vec<String>,
}

#[derive(Default)]
struct SheetOrderHistory {
    undo: Vec<SheetOrderEntry>,
    redo: Vec<SheetOrderEntry>,
}

enum HistoryAction {
    Push(SheetOrderEntry),
    Undo(SheetOrderEntry),
    Redo(SheetOrderEntry),
}

pub(crate) struct WorkbookAuthority {
    doc: Doc,
    base: Arc<WorkbookBase>,
    history: SheetOrderHistory,
    next_sheet_id: u64,
    undo_stack: Vec<StackItem<()>>,
    redo_stack: Vec<StackItem<()>>,
}

impl WorkbookAuthority {
    #[cfg(test)]
    fn from_model(model: &WorkbookModel) -> Result<Self, AuthorityError> {
        Self::from_model_internal(model, None, &[], None)
    }

    #[cfg(test)]
    fn from_model_with_client_id(
        model: &WorkbookModel,
        client_id: u64,
    ) -> Result<Self, AuthorityError> {
        Self::from_model_internal(model, Some(client_id), &[], None)
    }

    pub(crate) fn from_source(
        model: &WorkbookModel,
        client_id: Option<u64>,
        legacy_dimensions: &[xlsx_parse::LegacySheetDimensions],
        legacy_styles: Option<&Stylesheet>,
    ) -> Result<Self, AuthorityError> {
        Self::from_model_internal(model, client_id, legacy_dimensions, legacy_styles)
    }

    fn from_model_internal(
        model: &WorkbookModel,
        client_id: Option<u64>,
        legacy_dimensions: &[xlsx_parse::LegacySheetDimensions],
        legacy_styles: Option<&Stylesheet>,
    ) -> Result<Self, AuthorityError> {
        let base = WorkbookBase::from_model_with_legacy_dimensions(
            model,
            legacy_dimensions,
            legacy_styles,
        )
        .map_err(AuthorityError::InvalidState)?;
        if client_id == Some(base.bootstrap_client_id) {
            return Err(AuthorityError::ClientIdConflict(base.bootstrap_client_id));
        }

        let bootstrap = Doc::with_client_id(base.bootstrap_client_id);
        let keys = (0..model.sheets.len())
            .map(|index| format!("sheet:{index}"))
            .collect::<Vec<_>>();
        seed(&bootstrap, &base, model, &keys).map_err(AuthorityError::InvalidState)?;
        let bootstrap_update = bootstrap
            .transact()
            .encode_state_as_update_v1(&StateVector::default());

        let doc = match client_id {
            Some(client_id) => Doc::with_client_id(client_id),
            None => loop {
                let candidate = Doc::new();
                if candidate.client_id().get() != base.bootstrap_client_id {
                    break candidate;
                }
            },
        };
        hydrate_local_doc(&doc, &bootstrap_update).map_err(AuthorityError::InvalidState)?;
        let authority = Self {
            doc,
            base: Arc::new(base),
            history: SheetOrderHistory::default(),
            next_sheet_id: 0,
            undo_stack: Vec::new(),
            redo_stack: Vec::new(),
        };
        authority
            .strict_materialize()
            .map_err(AuthorityError::InvalidState)?;
        Ok(authority)
    }

    pub(crate) fn client_id(&self) -> u64 {
        self.doc.client_id().get()
    }

    pub(crate) fn state_vector_entries(&self) -> usize {
        self.doc.transact().state_vector().len()
    }

    pub(crate) fn materialize(&self) -> Result<WorkbookModel, AuthorityError> {
        self.materialize_internal(false)
            .map(|(model, _)| model)
            .map_err(AuthorityError::InvalidState)
    }

    pub(crate) fn structure(&self) -> Result<WorkbookStructure, AuthorityError> {
        self.materialize_internal(false)
            .map(|(_, structure)| structure)
            .map_err(AuthorityError::InvalidState)
    }

    /// Applies `ops`, whose style indices index `styles`, the caller's style table.
    pub(crate) fn apply_ops(
        &mut self,
        ops: &[Op],
        origin: SyncOrigin,
        styles: &Stylesheet,
    ) -> Result<Option<Vec<u8>>, AuthorityError> {
        let state_vector = self.doc.transact().state_vector();
        let mut model = self.materialize()?;
        let ops =
            remap_styles(ops, styles, &mut model.styles).map_err(AuthorityError::InvalidState)?;
        let ops = ops.as_ref();
        let authored_styles = ops
            .iter()
            .filter_map(|op| match op {
                Op::SetCell { sheet, at, .. } => Some((
                    (sheet.0, *at),
                    model
                        .sheet(*sheet)
                        .and_then(|sheet| sheet.cell(*at))
                        .and_then(|cell| cell.style),
                )),
                _ => None,
            })
            .collect::<HashMap<_, _>>();
        for op in ops {
            xlsx_ops::apply_in_place(&mut model, op).map_err(|error| {
                AuthorityError::InvalidState(format!(
                    "cannot apply local operation to authored state: {error}"
                ))
            })?;
        }
        if self.base.defined_names != model.defined_names {
            Arc::make_mut(&mut self.base).defined_names = model.defined_names.clone();
        }
        self.sync_model(&model, ops, origin, &authored_styles)
            .map_err(AuthorityError::InvalidState)?;
        let update = self.doc.transact().encode_diff_v1(&state_vector);
        Ok((update.as_slice() != Update::EMPTY_V1).then_some(update))
    }

    pub(crate) fn encode_state_vector_v1(&self) -> Vec<u8> {
        let state_vector = self.doc.transact().state_vector();
        let mut entries = state_vector
            .iter()
            .map(|(client, clock)| (client.get(), *clock))
            .collect::<Vec<_>>();
        entries.sort_unstable();
        let mut encoder = EncoderV1::new();
        encoder.write_var(entries.len());
        for (client, clock) in entries {
            encoder.write_var(client);
            encoder.write_var(clock);
        }
        encoder.to_vec()
    }

    pub(crate) fn encode_state_as_update_v1(&self) -> Vec<u8> {
        self.doc
            .transact()
            .encode_state_as_update_v1(&StateVector::default())
    }

    pub(crate) fn encode_diff_v1(
        &self,
        remote_state_vector: &[u8],
    ) -> Result<Vec<u8>, AuthorityError> {
        let state_vector = decode_state_vector_v1(remote_state_vector)
            .map_err(AuthorityError::InvalidStateVector)?;
        Ok(self.doc.transact().encode_diff_v1(&state_vector))
    }

    /// The replica this one would become by taking a persisted snapshot as its
    /// whole state and upgrading it to the current schema.
    ///
    /// Merging cannot restore a snapshot an earlier release wrote. The
    /// bootstrap client ID is the head of the base fingerprint, which hashes
    /// the schema version, so a bootstrap this build seeds never dedupes
    /// against the one such a snapshot carries: the two bases double up and
    /// the survivor is whichever client ID sorts higher. A snapshot supersedes
    /// an untouched bootstrap of the same workbook whatever schema it was
    /// written at, and where the bootstraps do agree the two are the same
    /// document, so adopting is never worse than merging.
    pub(crate) fn snapshot_replacement(&self, update: &[u8]) -> SnapshotAdoption {
        if !self.is_pristine() {
            return SnapshotAdoption::NotApplicable;
        }
        let doc = Doc::with_client_id(self.client_id());
        if hydrate_doc(&doc, update).is_err() {
            return SnapshotAdoption::NotApplicable;
        }
        let candidate = Self {
            doc,
            base: self.base.clone(),
            history: SheetOrderHistory::default(),
            next_sheet_id: self.next_sheet_id,
            undo_stack: Vec::new(),
            redo_stack: Vec::new(),
        };
        if !candidate.is_whole_document() {
            return SnapshotAdoption::NotApplicable;
        }
        if let Err(error) = candidate.upgrade_schema() {
            return SnapshotAdoption::Incompatible(error);
        }
        match candidate.strict_materialize() {
            Err(error) => SnapshotAdoption::Incompatible(error),
            Ok(_) => SnapshotAdoption::Replacement(Box::new(candidate)),
        }
    }

    /// True while the replica still holds nothing but its own bootstrap.
    fn is_pristine(&self) -> bool {
        let txn = self.doc.transact();
        let state_vector = txn.state_vector();
        state_vector.len() == 1
            && state_vector
                .iter()
                .all(|(client, _)| client.get() == self.base.bootstrap_client_id)
            && txn.snapshot().delete_set.is_empty()
    }

    /// True when the document stands on its own rather than being the tail of
    /// someone else's — an incremental update hydrates into neither the roots
    /// nor the metadata a whole workbook carries.
    fn is_whole_document(&self) -> bool {
        let txn = self.doc.transact();
        if require_root_keys(&txn, &[CELL_FORMATS, META, SHEET_ORDER, SHEETS]).is_err() {
            return false;
        }
        txn.get_map(META).is_some_and(|meta| {
            require_map_keys(
                &meta,
                &txn,
                &[BASE_FINGERPRINT, "schemaVersion", STRUCTURE_GENERATION],
                "workbook metadata",
            )
            .is_ok()
        })
    }

    /// `baseline` may carry this replica's already-encoded current state.
    pub(crate) fn stage_updates_v1(
        &self,
        updates: &[&[u8]],
        baseline: Option<&[u8]>,
    ) -> Result<StagedUpdate, AuthorityError> {
        if updates.is_empty() {
            return Err(AuthorityError::InvalidUpdate(
                "no updates were provided".to_string(),
            ));
        }
        let decoded = updates
            .iter()
            .map(|update| decode_update_v1(update).map_err(AuthorityError::InvalidUpdate))
            .collect::<Result<Vec<_>, _>>()?;
        let incoming = if updates.len() == 1 {
            decoded.into_iter().next().unwrap()
        } else {
            Update::merge_updates(decoded)
        };
        let before_vector = self.doc.transact().state_vector();
        let owned;
        let before = match baseline {
            Some(baseline) => baseline,
            None => {
                owned = self.encode_state_as_update_v1();
                &owned
            }
        };
        let staged_doc = Doc::with_client_id(self.client_id());
        hydrate_local_doc(&staged_doc, before).map_err(AuthorityError::InvalidState)?;
        staged_doc
            .transact_mut_with(REMOTE_ORIGIN)
            .apply_update(incoming)
            .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))?;

        let staged = Self {
            doc: staged_doc,
            base: self.base.clone(),
            history: SheetOrderHistory::default(),
            next_sheet_id: self.next_sheet_id,
            undo_stack: Vec::new(),
            redo_stack: Vec::new(),
        };
        let pending = {
            let txn = staged.doc.transact();
            txn.store().pending_update().is_some() || txn.store().pending_ds().is_some()
        };
        if let Err(error) = staged.upgrade_schema()
            && !pending
        {
            return Err(AuthorityError::InvalidState(error));
        }
        let after = staged.encode_state_as_update_v1();
        let integrated = staged.doc.transact().encode_diff_v1(&before_vector);
        let after_vector = staged.doc.transact().state_vector();
        let state_vector_entries = after_vector.len();
        // The current state only needs materializing when the state vectors
        // match: a differing vector already proves the update changed
        // something, while an equal one can still hide a delete-set change, so
        // the models must actually be compared there.
        if pending {
            let mut current = None;
            if integrated.as_slice() != Update::EMPTY_V1
                && let Ok((model, structure)) = staged.strict_materialize()
            {
                let effective = after_vector != before_vector || {
                    let state = self
                        .strict_materialize()
                        .map_err(AuthorityError::InvalidState)?;
                    let effective = model != state.0 || structure != state.1;
                    current = Some(state);
                    effective
                };
                if effective {
                    return Ok(StagedUpdate {
                        commit_update: integrated.clone(),
                        effective: true,
                        model,
                        pending: true,
                        state_bytes: after.len(),
                        state_vector_entries,
                        structure,
                        update: integrated,
                    });
                }
            }
            let (current_model, current_structure) = match current {
                Some(state) => state,
                None => self
                    .strict_materialize()
                    .map_err(AuthorityError::InvalidState)?,
            };
            return Ok(StagedUpdate {
                commit_update: Update::EMPTY_V1.to_vec(),
                effective: false,
                model: current_model,
                pending: true,
                state_bytes: before.len(),
                state_vector_entries: self.state_vector_entries(),
                structure: current_structure,
                update: Update::EMPTY_V1.to_vec(),
            });
        }
        let (model, structure) = staged
            .strict_materialize()
            .map_err(AuthorityError::InvalidState)?;
        let effective = after_vector != before_vector || {
            let (current_model, current_structure) = self
                .strict_materialize()
                .map_err(AuthorityError::InvalidState)?;
            model != current_model || structure != current_structure
        };
        Ok(StagedUpdate {
            commit_update: integrated.clone(),
            effective,
            model,
            pending,
            state_bytes: after.len(),
            state_vector_entries,
            structure,
            update: integrated,
        })
    }

    pub(crate) fn stage_local_ops_v1(
        &self,
        ops: &[Op],
        origin: SyncOrigin,
        styles: &Stylesheet,
    ) -> Result<StagedLocalUpdate, AuthorityError> {
        self.stage_local_ops_from_v1(ops, origin, styles, &self.encode_state_as_update_v1())
    }

    /// [`Self::stage_local_ops_v1`] from this replica's already-encoded state.
    pub(crate) fn stage_local_ops_from_v1(
        &self,
        ops: &[Op],
        origin: SyncOrigin,
        styles: &Stylesheet,
        baseline: &[u8],
    ) -> Result<StagedLocalUpdate, AuthorityError> {
        let staged_doc = Doc::with_client_id(self.client_id());
        hydrate_local_doc(&staged_doc, baseline).map_err(AuthorityError::InvalidState)?;
        let mut staged = Self {
            doc: staged_doc,
            base: self.base.clone(),
            history: SheetOrderHistory::default(),
            next_sheet_id: self.next_sheet_id,
            undo_stack: Vec::new(),
            redo_stack: Vec::new(),
        };
        // `apply_ops` already encoded the same diff: the staged doc is the
        // hydrated baseline, so its pre-op state vector is this replica's own.
        let update = staged
            .apply_ops(ops, origin, styles)?
            .unwrap_or_else(|| Update::EMPTY_V1.to_vec());
        let state = staged.encode_state_as_update_v1();
        let state_vector_entries = staged.doc.transact().state_vector().len();
        let (model, structure) = staged
            .materialize_internal(false)
            .map_err(AuthorityError::InvalidState)?;
        Ok(StagedLocalUpdate {
            model,
            state_bytes: state.len(),
            state_vector_entries,
            structure,
            update,
            carried: CarriedState {
                sheet_order: staged.history.undo.pop(),
                base: staged.base,
                next_sheet_id: staged.next_sheet_id,
            },
        })
    }

    /// Adopts a staged local update, returning its bytes unless it was empty. They are decoded
    /// before anything changes, so a failure leaves this replica as it was.
    pub(crate) fn adopt_local_update(
        &mut self,
        staged: StagedLocalUpdate,
        history: LocalHistory,
    ) -> Result<Option<Vec<u8>>, AuthorityError> {
        let StagedLocalUpdate {
            update: bytes,
            carried,
            ..
        } = staged;
        let update = decode_local_update_v1(&bytes).map_err(AuthorityError::InvalidUpdate)?;
        let undo = match history {
            LocalHistory::Undo => Some(
                build_undo_manager(&self.doc, self.undo_stack.clone(), self.redo_stack.clone())
                    .map_err(AuthorityError::InvalidState)?,
            ),
            LocalHistory::SheetOrder | LocalHistory::None => None,
        };
        match undo {
            Some(mut undo) => {
                undo.reset();
                self.doc
                    .transact_mut_with(self.client_id())
                    .apply_update(update)
                    .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))?;
                self.undo_stack = undo.undo_stack().to_vec();
                self.redo_stack = undo.redo_stack().to_vec();
            }
            None => self
                .doc
                .transact_mut_with(UNTRACKED_ORIGIN)
                .apply_update(update)
                .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))?,
        }
        let CarriedState {
            base,
            next_sheet_id,
            sheet_order,
        } = carried;
        self.base = base;
        self.next_sheet_id = next_sheet_id;
        if history == LocalHistory::SheetOrder
            && let Some(entry) = sheet_order
        {
            self.apply_history(HistoryAction::Push(entry));
        }
        Ok((bytes.as_slice() != Update::EMPTY_V1).then_some(bytes))
    }

    /// Whether `update` integrates completely into a fresh replica of `baseline`, this
    /// replica's encoded state.
    pub(crate) fn rehearse_local_update(
        &self,
        baseline: &[u8],
        update: &[u8],
    ) -> Result<bool, AuthorityError> {
        let doc = Doc::with_client_id(self.client_id());
        hydrate_local_doc(&doc, baseline).map_err(AuthorityError::InvalidState)?;
        let update = decode_local_update_v1(update).map_err(AuthorityError::InvalidUpdate)?;
        doc.transact_mut_with(UNTRACKED_ORIGIN)
            .apply_update(update)
            .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))?;
        Ok(!has_pending(&doc))
    }

    /// Whether this replica holds update structs it has not integrated yet.
    pub(crate) fn has_pending_updates(&self) -> bool {
        has_pending(&self.doc)
    }

    pub(crate) fn apply_local_update_v1(
        &mut self,
        update: &[u8],
        origin: SyncOrigin,
    ) -> Result<(), AuthorityError> {
        let update = decode_local_update_v1(update).map_err(AuthorityError::InvalidUpdate)?;
        if origin == SyncOrigin::User {
            let mut undo =
                build_undo_manager(&self.doc, self.undo_stack.clone(), self.redo_stack.clone())
                    .map_err(AuthorityError::InvalidState)?;
            undo.reset();
            self.doc
                .transact_mut_with(self.client_id())
                .apply_update(update)
                .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))?;
            self.undo_stack = undo.undo_stack().to_vec();
            self.redo_stack = undo.redo_stack().to_vec();
            Ok(())
        } else {
            self.doc
                .transact_mut_with(origin.as_str())
                .apply_update(update)
                .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))
        }
    }

    pub(crate) fn apply_staged_update_v1(&self, update: &[u8]) -> Result<(), AuthorityError> {
        let update = decode_local_update_v1(update).map_err(AuthorityError::InvalidUpdate)?;
        self.doc
            .transact_mut_with(REMOTE_ORIGIN)
            .apply_update(update)
            .map_err(|error| AuthorityError::InvalidUpdate(error.to_string()))
    }

    pub(crate) fn can_undo(&self) -> bool {
        !self.undo_stack.is_empty()
    }

    pub(crate) fn can_redo(&self) -> bool {
        !self.redo_stack.is_empty()
    }

    pub(crate) fn undo_depth(&self) -> usize {
        self.undo_stack.len()
    }

    pub(crate) fn redo_depth(&self) -> usize {
        self.redo_stack.len()
    }

    /// The document and history as they stand, so a caller that rejects what a
    /// history step produced can put the authority back. The step itself is
    /// only half the change: the caller validates the other half.
    pub(crate) fn checkpoint(&self) -> AuthorityCheckpoint {
        AuthorityCheckpoint {
            state: self.encode_state_as_update_v1(),
            undo_stack: self.undo_stack.clone(),
            redo_stack: self.redo_stack.clone(),
        }
    }

    pub(crate) fn restore(
        &mut self,
        checkpoint: AuthorityCheckpoint,
    ) -> Result<(), AuthorityError> {
        self.rollback(
            &checkpoint.state,
            checkpoint.undo_stack,
            checkpoint.redo_stack,
        )
    }

    pub(crate) fn undo(&mut self) -> Result<Option<HistoryUpdate>, AuthorityError> {
        self.apply_history_change(false)
    }

    pub(crate) fn redo(&mut self) -> Result<Option<HistoryUpdate>, AuthorityError> {
        self.apply_history_change(true)
    }

    /// Moves the document and both stacks. Every failure here leaves that half
    /// done, so the caller holds a checkpoint and puts it back — one checkpoint
    /// for the whole step rather than one here and another there, since this
    /// runs on a keystroke and each costs a copy of the document.
    fn apply_history_change(
        &mut self,
        redo: bool,
    ) -> Result<Option<HistoryUpdate>, AuthorityError> {
        let state_vector = self.doc.transact().state_vector();
        let mut undo =
            build_undo_manager(&self.doc, self.undo_stack.clone(), self.redo_stack.clone())
                .map_err(AuthorityError::InvalidState)?;
        let applied = if redo {
            undo.redo_blocking()
        } else {
            undo.undo_blocking()
        };
        self.undo_stack = undo.undo_stack().to_vec();
        self.redo_stack = undo.redo_stack().to_vec();
        drop(undo);
        if !applied {
            return Ok(None);
        }
        let (model, structure) = self
            .strict_materialize()
            .map_err(AuthorityError::InvalidState)?;
        let update = self.doc.transact().encode_diff_v1(&state_vector);
        Ok(Some(HistoryUpdate {
            model,
            structure,
            update,
        }))
    }

    /// Rebuilds the document from a checkpoint. The GUID is carried over
    /// because a history entry names the document it came from: a fresh one
    /// orphans every restored entry, and the undo manager discards orphans
    /// silently, one whole stack at a time.
    fn rollback(
        &mut self,
        restore: &[u8],
        undo_stack: Vec<StackItem<()>>,
        redo_stack: Vec<StackItem<()>>,
    ) -> Result<(), AuthorityError> {
        let doc = Doc::with_options(Options::with_guid_and_client_id(
            self.doc.guid().clone(),
            self.doc.client_id(),
        ));
        hydrate_local_doc(&doc, restore).map_err(AuthorityError::InvalidState)?;
        self.doc = doc;
        self.undo_stack = undo_stack;
        self.redo_stack = redo_stack;
        Ok(())
    }

    pub(crate) fn clear_history(&mut self) {
        self.history = SheetOrderHistory::default();
    }

    fn strict_materialize(&self) -> Result<(WorkbookModel, WorkbookStructure), String> {
        self.materialize_internal(true)
    }

    fn upgrade_schema(&self) -> Result<bool, String> {
        let version = self.schema_version()?;
        validate_schema_version(version)?;
        self.deduplicate_sheet_order()?;
        if version == SCHEMA_VERSION && self.has_current_base_fingerprint() {
            return Ok(false);
        }
        let (model, structure) = self.materialize_internal(false)?;
        let features = structure
            .sheet_keys
            .iter()
            .zip(&model.sheets)
            .map(|(key, sheet)| {
                let hyperlinks = serde_json::to_string(&sheet.hyperlinks)
                    .map_err(|error| format!("cannot encode sheet hyperlinks: {error}"))?;
                let charts = serde_json::to_string(&sheet.charts)
                    .map_err(|error| format!("cannot encode sheet charts: {error}"))?;
                Ok((
                    key.clone(),
                    (
                        sheet.freeze_pane,
                        hyperlinks,
                        charts,
                        sheet.col_widths.clone(),
                        sheet.row_heights.clone(),
                    ),
                ))
            })
            .collect::<Result<BTreeMap<_, _>, String>>()?;
        let mut txn = self.doc.transact_mut_with(HYDRATE_ORIGIN);
        let sheets = txn
            .get_map(SHEETS)
            .ok_or_else(|| "missing sheet map".to_string())?;
        let keys = sheets.keys(&txn).map(str::to_string).collect::<Vec<_>>();
        for key in keys {
            let sheet = sheets
                .get(&txn, &key)
                .and_then(|value| value.cast::<MapRef>().ok())
                .ok_or_else(|| format!("sheet {key} is not a map"))?;
            if let Some((.., col_widths, row_heights)) = features.get(&key) {
                let widths: MapRef = sheet.get_or_init(&mut txn, COL_WIDTHS);
                sync_numbers(&widths, &mut txn, col_widths);
                let heights: MapRef = sheet.get_or_init(&mut txn, ROW_HEIGHTS);
                sync_numbers(&heights, &mut txn, row_heights);
            }
            let (freeze_pane, hyperlinks, charts) = features
                .get(&key)
                .map(|(freeze_pane, hyperlinks, charts, ..)| {
                    (*freeze_pane, hyperlinks.as_str(), charts.as_str())
                })
                .unwrap_or((None, "[]", "[]"));
            if version < FREEZE_PANE_SCHEMA_VERSION {
                sheet.try_update(&mut txn, FREEZE_PANE, freeze_pane_to_any(freeze_pane));
            }
            if version < HYPERLINK_SCHEMA_VERSION {
                sheet.try_update(&mut txn, HYPERLINKS, hyperlinks);
            }
            if version < CHARTS_SCHEMA_VERSION {
                sheet.try_update(&mut txn, CHARTS, charts);
            }
        }
        let meta = txn
            .get_map(META)
            .ok_or_else(|| "missing workbook metadata".to_string())?;
        meta.try_update(&mut txn, BASE_FINGERPRINT, self.base.fingerprint.as_str());
        meta.try_update(&mut txn, "schemaVersion", SCHEMA_VERSION);
        Ok(true)
    }

    fn deduplicate_sheet_order(&self) -> Result<(), String> {
        let mut txn = self.doc.transact_mut_with(HYDRATE_ORIGIN);
        let order = txn
            .get_array(SHEET_ORDER)
            .ok_or_else(|| "missing sheet order".to_string())?;
        let keys = sheet_keys(&order, &txn)?;
        let mut seen = HashSet::with_capacity(keys.len());
        let duplicates = keys
            .iter()
            .enumerate()
            .filter_map(|(index, key)| (!seen.insert(key)).then_some(index))
            .collect::<Vec<_>>();
        for index in duplicates.into_iter().rev() {
            order.remove(&mut txn, yrs_index(index)?);
        }
        Ok(())
    }

    fn schema_version(&self) -> Result<i64, String> {
        let txn = self.doc.transact();
        txn.get_map(META)
            .and_then(|meta| meta.get(&txn, "schemaVersion"))
            .and_then(|value| value.cast::<i64>().ok())
            .ok_or_else(|| "missing schema version".to_string())
    }

    fn has_current_base_fingerprint(&self) -> bool {
        let txn = self.doc.transact();
        txn.get_map(META)
            .and_then(|meta| meta.get(&txn, BASE_FINGERPRINT))
            .and_then(|value| value.cast::<String>().ok())
            .is_some_and(|fingerprint| fingerprint == self.base.fingerprint)
    }

    fn materialize_internal(
        &self,
        strict: bool,
    ) -> Result<(WorkbookModel, WorkbookStructure), String> {
        let txn = self.doc.transact();
        if strict {
            require_root_keys(&txn, &[CELL_FORMATS, META, SHEET_ORDER, SHEETS])?;
        }
        let meta = txn
            .get_map(META)
            .ok_or_else(|| "missing workbook metadata".to_string())?;
        if strict {
            require_map_keys(
                &meta,
                &txn,
                &[BASE_FINGERPRINT, "schemaVersion", STRUCTURE_GENERATION],
                "workbook metadata",
            )?;
        }
        let version = meta
            .get(&txn, "schemaVersion")
            .and_then(|value| value.cast::<i64>().ok())
            .ok_or_else(|| "missing schema version".to_string())?;
        validate_schema_version(version)?;
        let fingerprint = meta
            .get(&txn, BASE_FINGERPRINT)
            .and_then(|value| value.cast::<String>().ok())
            .ok_or_else(|| "missing workbook base fingerprint".to_string())?;
        if !self.base.accepts_fingerprint(version, &fingerprint) {
            return Err("workbook base fingerprint does not match shared state".to_string());
        }
        let generation = structure_generation(&meta, &txn)?;
        let cell_formats = txn
            .get_map(CELL_FORMATS)
            .ok_or_else(|| "missing cell format catalog".to_string())?;
        let (styles, style_indices) =
            materialize_cell_formats(&cell_formats, &txn, &self.base.styles)?;

        let order = txn
            .get_array(SHEET_ORDER)
            .ok_or_else(|| "missing sheet order".to_string())?;
        let sheets = txn
            .get_map(SHEETS)
            .ok_or_else(|| "missing sheet map".to_string())?;
        let keys = sheet_keys(&order, &txn)?;
        let mut seen = HashSet::with_capacity(keys.len());
        let mut model = self.base.workbook();
        model.styles = styles;
        let expected_sheet_keys = sheet_schema_keys(version);
        let optional_sheet_keys = sheet_schema_optional_keys(version);
        for key in keys.iter() {
            if !seen.insert(key.clone()) {
                return Err(format!("duplicate sheet key {key}"));
            }
            let sheet_map = sheets
                .get(&txn, key)
                .and_then(|value| value.cast::<MapRef>().ok())
                .ok_or_else(|| format!("missing sheet {key}"))?;
            if strict {
                require_map_keys_with_optional(
                    &sheet_map,
                    &txn,
                    expected_sheet_keys,
                    optional_sheet_keys,
                    &format!("sheet {key}"),
                )?;
            }
            let base_sheet = base_sheet_index(key);
            let freeze_pane = base_sheet
                .and_then(|base| self.base.freeze_panes.get(base))
                .copied()
                .flatten();
            let format = base_sheet
                .and_then(|base| self.base.formats.get(base))
                .copied()
                .unwrap_or_default();
            let col_styles = base_sheet
                .and_then(|base| self.base.col_styles.get(base))
                .map(Vec::as_slice)
                .unwrap_or_default();
            let hyperlinks = base_sheet
                .and_then(|base| self.base.hyperlinks.get(base))
                .map(Vec::as_slice)
                .unwrap_or_default();
            let charts = base_sheet
                .and_then(|base| self.base.charts.get(base))
                .map(Vec::as_slice)
                .unwrap_or_default();
            let hidden_dimensions = base_sheet
                .and_then(|base| self.base.hidden_dimensions.get(base))
                .unwrap_or(&EMPTY_HIDDEN_DIMENSIONS);
            model.sheets.push(materialize_sheet(
                &sheet_map,
                &txn,
                &style_indices,
                version,
                SheetFallbacks {
                    freeze_pane,
                    format,
                    col_styles,
                    hyperlinks,
                    charts,
                    hidden_dimensions,
                },
            )?);
        }
        project_shared_frame_anchors(&mut model.sheets);
        let active = keys.iter().cloned().collect::<BTreeSet<_>>();
        let all_keys = sheets
            .keys(&txn)
            .map(str::to_string)
            .collect::<BTreeSet<_>>();
        let mut shared_types = BTreeMap::new();
        for key in all_keys {
            let sheet_map = sheets
                .get(&txn, &key)
                .and_then(|value| value.cast::<MapRef>().ok())
                .ok_or_else(|| format!("sheet {key} is not a map"))?;
            if strict && !active.contains(&key) {
                require_map_keys_with_optional(
                    &sheet_map,
                    &txn,
                    expected_sheet_keys,
                    optional_sheet_keys,
                    &format!("inactive sheet {key}"),
                )?;
                materialize_sheet(
                    &sheet_map,
                    &txn,
                    &style_indices,
                    version,
                    SheetFallbacks::default(),
                )?;
            }
            shared_types.insert(key, sheet_shared_types(&sheet_map, &txn)?);
        }
        let structure = WorkbookStructure {
            generation,
            sheet_keys: keys,
            sheet_names: model
                .sheets
                .iter()
                .map(|sheet| sheet.name.clone())
                .collect(),
            freeze_panes: model.sheets.iter().map(|sheet| sheet.freeze_pane).collect(),
            hyperlinks: model
                .sheets
                .iter()
                .map(|sheet| sheet.hyperlinks.clone())
                .collect(),
            charts: model
                .sheets
                .iter()
                .map(|sheet| sheet.charts.iter().map(ChartIdentity::of).collect())
                .collect(),
            merges: model
                .sheets
                .iter()
                .map(|sheet| sheet.merges.clone())
                .collect(),
            shared_types,
        };
        Ok((model, structure))
    }

    /// `authored_styles` holds the pre-batch style of every `SetCell` target.
    fn sync_model(
        &mut self,
        model: &WorkbookModel,
        ops: &[Op],
        origin: SyncOrigin,
        authored_styles: &HashMap<(u32, CellRef), Option<u32>>,
    ) -> Result<(), String> {
        let current_keys = self.current_sheet_keys()?;
        let (keys, history) =
            self.plan_sheet_keys(&current_keys, ops, model.sheets.len(), origin)?;
        self.validate_sync_state(&current_keys, &keys)?;

        let topology_changed = current_keys != keys;
        // a SetCell after AddSheet targets the post-insert index, so its
        // pre-batch baseline lookup would read the wrong sheet.
        let full_sync = ops.iter().any(requires_full_semantic_sync)
            || (ops.iter().any(|op| matches!(op, Op::AddSheet { .. }))
                && ops.iter().any(|op| matches!(op, Op::SetCell { .. })));
        let structure_delta = i64::try_from(ops.iter().filter(|op| is_structural_op(op)).count())
            .map_err(|_| "too many structural operations".to_string())?;
        let mut authored_cells = HashSet::new();
        let mut formatted_cells = HashSet::new();
        let mut col_widths = HashSet::new();
        let mut row_heights = HashSet::new();
        let mut merges = HashSet::new();
        if !full_sync {
            let targets = targeted_sheet_keys(&current_keys, &keys, ops)?;
            for (op, target) in ops.iter().zip(targets) {
                match (op, target) {
                    (Op::SetCell { sheet, at, cell }, Some(key)) => {
                        let current_style = authored_styles.get(&(sheet.0, *at)).copied().flatten();
                        if cell.style != current_style {
                            formatted_cells.insert((key.clone(), *at));
                        }
                        authored_cells.insert((key, *at));
                    }
                    (Op::SetColWidth { col, .. }, Some(key)) => {
                        col_widths.insert((key, *col));
                    }
                    (Op::SetRowHeight { row, .. }, Some(key)) => {
                        row_heights.insert((key, *row));
                    }
                    (Op::MergeCells { .. } | Op::UnmergeCells { .. }, Some(key)) => {
                        merges.insert(key);
                    }
                    (
                        Op::PatchRangeStyle { range, .. }
                        | Op::SetRangeNumberFormat { range, .. }
                        | Op::ApplyRangeFormat { range, .. },
                        Some(key),
                    ) => {
                        for row in range.start.row..=range.end.row {
                            for col in range.start.col..=range.end.col {
                                formatted_cells.insert((key.clone(), CellRef::new(row, col)));
                            }
                        }
                    }
                    (Op::AddSheet { .. } | Op::RemoveSheet { .. }, None) => {}
                    (_, None) => {}
                    _ => return Err("semantic operation requires a full sync".to_string()),
                }
            }
        }
        let current_key_set = current_keys
            .iter()
            .map(String::as_str)
            .collect::<HashSet<_>>();
        let newly_active = keys
            .iter()
            .filter(|key| !current_key_set.contains(key.as_str()))
            .cloned()
            .collect::<HashSet<_>>();
        if !topology_changed
            && !full_sync
            && structure_delta == 0
            && authored_cells.is_empty()
            && formatted_cells.is_empty()
            && col_widths.is_empty()
            && row_heights.is_empty()
            && merges.is_empty()
        {
            self.apply_history(history);
            return Ok(());
        }

        let mut txn = self.doc.transact_mut_with(origin.as_str());
        let cell_formats = txn
            .get_map(CELL_FORMATS)
            .ok_or_else(|| "missing cell format catalog".to_string())?;
        sync_cell_formats(&cell_formats, &mut txn, &model.styles)?;
        let order = txn
            .get_array(SHEET_ORDER)
            .ok_or_else(|| "missing sheet order".to_string())?;
        let sheets = txn
            .get_map(SHEETS)
            .ok_or_else(|| "missing sheet map".to_string())?;
        if structure_delta != 0 {
            let meta = txn
                .get_map(META)
                .ok_or_else(|| "missing workbook metadata".to_string())?;
            let next = structure_generation(&meta, &txn)?
                .checked_add(structure_delta)
                .ok_or_else(|| "structure generation overflow".to_string())?;
            meta.try_update(&mut txn, STRUCTURE_GENERATION, next);
        }
        if topology_changed {
            patch_sheet_order(&order, &mut txn, &current_keys, &keys)?;
        }
        if full_sync {
            for (key, sheet) in keys.iter().zip(&model.sheets) {
                let sheet_map = sheet_map_for_sync(&sheets, &mut txn, key)?;
                sync_sheet(&sheet_map, &mut txn, sheet, &model.styles)?;
            }
        } else {
            for (key, sheet) in keys.iter().zip(&model.sheets) {
                if newly_active.contains(key) {
                    let sheet_map = sheet_map_for_sync(&sheets, &mut txn, key)?;
                    sync_sheet(&sheet_map, &mut txn, sheet, &model.styles)?;
                }
            }
            for (key, at) in authored_cells {
                let (sheet_map, sheet_model) =
                    sheet_parts_by_key(&sheets, &txn, &keys, model, &key)?;
                sync_authored_cell(&sheet_map, &mut txn, sheet_model, at)?;
            }
            for (key, at) in formatted_cells {
                let (sheet_map, sheet_model) =
                    sheet_parts_by_key(&sheets, &txn, &keys, model, &key)?;
                sync_cell_format(&sheet_map, &mut txn, sheet_model, &model.styles, at)?;
            }
            for (key, col) in col_widths {
                let (sheet_map, sheet_model) =
                    sheet_parts_by_key(&sheets, &txn, &keys, model, &key)?;
                let map: MapRef = sheet_map.get_or_init(&mut txn, COL_WIDTHS);
                sync_number(
                    &map,
                    &mut txn,
                    col,
                    sheet_model.col_widths.get(&col).copied(),
                );
            }
            for (key, row) in row_heights {
                let (sheet_map, sheet_model) =
                    sheet_parts_by_key(&sheets, &txn, &keys, model, &key)?;
                let map: MapRef = sheet_map.get_or_init(&mut txn, ROW_HEIGHTS);
                sync_number(
                    &map,
                    &mut txn,
                    row,
                    sheet_model.row_heights.get(&row).copied(),
                );
            }
            for key in merges {
                let (sheet_map, sheet_model) =
                    sheet_parts_by_key(&sheets, &txn, &keys, model, &key)?;
                sheet_map.try_update(&mut txn, MERGES, merges_to_any(&sheet_model.merges));
            }
        }
        drop(txn);
        self.apply_history(history);
        Ok(())
    }

    fn current_sheet_keys(&self) -> Result<Vec<String>, String> {
        let txn = self.doc.transact();
        let order = txn
            .get_array(SHEET_ORDER)
            .ok_or_else(|| "missing sheet order".to_string())?;
        sheet_keys(&order, &txn)
    }

    fn validate_sync_state(&self, current: &[String], desired: &[String]) -> Result<(), String> {
        let txn = self.doc.transact();
        let sheets = txn
            .get_map(SHEETS)
            .ok_or_else(|| "missing sheet map".to_string())?;
        for key in current {
            match sheets.get(&txn, key) {
                Some(Out::YMap(_)) => {}
                Some(_) => return Err(format!("sheet {key} is not a map")),
                None => return Err(format!("missing sheet {key}")),
            }
        }
        for key in desired {
            if let Some(value) = sheets.get(&txn, key)
                && !matches!(value, Out::YMap(_))
            {
                return Err(format!("sheet {key} is not a map"));
            }
        }
        Ok(())
    }

    fn plan_sheet_keys(
        &mut self,
        current: &[String],
        ops: &[Op],
        final_len: usize,
        origin: SyncOrigin,
    ) -> Result<(Vec<String>, HistoryAction), String> {
        match origin {
            SyncOrigin::User | SyncOrigin::Agent => {
                let keys = self.reconcile_sheet_keys(current.to_vec(), ops, final_len)?;
                let entry = SheetOrderEntry {
                    before: current.to_vec(),
                    after: keys.clone(),
                };
                Ok((keys, HistoryAction::Push(entry)))
            }
            SyncOrigin::Undo => {
                let entry = self
                    .history
                    .undo
                    .last()
                    .cloned()
                    .ok_or_else(|| "sheet-order undo history is empty".to_string())?;
                if entry.after != current {
                    return Err("sheet-order undo history does not match current state".to_string());
                }
                if entry.before.len() != final_len {
                    return Err("sheet-order undo result does not match workbook".to_string());
                }
                Ok((entry.before.clone(), HistoryAction::Undo(entry)))
            }
            SyncOrigin::Redo => {
                let entry = self
                    .history
                    .redo
                    .last()
                    .cloned()
                    .ok_or_else(|| "sheet-order redo history is empty".to_string())?;
                if entry.before != current {
                    return Err("sheet-order redo history does not match current state".to_string());
                }
                if entry.after.len() != final_len {
                    return Err("sheet-order redo result does not match workbook".to_string());
                }
                Ok((entry.after.clone(), HistoryAction::Redo(entry)))
            }
        }
    }

    fn reconcile_sheet_keys(
        &mut self,
        mut keys: Vec<String>,
        ops: &[Op],
        final_len: usize,
    ) -> Result<Vec<String>, String> {
        for op in ops {
            match op {
                Op::AddSheet { index, .. } => {
                    if *index > keys.len() {
                        return Err(format!("sheet insertion index {index} is out of range"));
                    }
                    let key = self.allocate_sheet_key();
                    keys.insert(*index, key);
                }
                Op::RemoveSheet { index } => {
                    if *index >= keys.len() {
                        return Err(format!("sheet removal index {index} is out of range"));
                    }
                    keys.remove(*index);
                }
                _ => {}
            }
        }
        if keys.len() != final_len {
            return Err("sheet order does not match workbook projection".to_string());
        }
        Ok(keys)
    }

    fn apply_history(&mut self, action: HistoryAction) {
        match action {
            HistoryAction::Push(entry) => {
                self.history.undo.push(entry);
                self.history.redo.clear();
            }
            HistoryAction::Undo(entry) => {
                self.history.undo.pop();
                self.history.redo.push(entry);
            }
            HistoryAction::Redo(entry) => {
                self.history.redo.pop();
                self.history.undo.push(entry);
            }
        }
    }

    fn allocate_sheet_key(&mut self) -> String {
        let key = format!("replica:{}:{}", self.client_id(), self.next_sheet_id);
        self.next_sheet_id += 1;
        key
    }
}

/// The base-model sheet a stable key names. Bootstrap mints `sheet:N` from the
/// base order and a replica mints `replica:...`, so a legacy state whose sheets
/// were reordered still reads its own fallback features rather than whichever
/// sheet now sits at that position.
fn base_sheet_index(key: &str) -> Option<usize> {
    key.strip_prefix("sheet:")?.parse().ok()
}

pub(crate) fn is_structural_op(op: &Op) -> bool {
    matches!(
        op,
        Op::InsertRows { .. }
            | Op::DeleteRows { .. }
            | Op::InsertCols { .. }
            | Op::DeleteCols { .. }
            | Op::SetFreezePane { .. }
            | Op::SetHyperlinks { .. }
            | Op::RestoreColStyles { .. }
            | Op::MergeCells { .. }
            | Op::UnmergeCells { .. }
            | Op::AddSheet { .. }
            | Op::RemoveSheet { .. }
            | Op::RenameSheet { .. }
            | Op::RestoreSheet { .. }
            | Op::SetCharts { .. }
            | Op::SetDefinedNames { .. }
    )
}

fn seed(
    doc: &Doc,
    base: &WorkbookBase,
    model: &WorkbookModel,
    keys: &[String],
) -> Result<(), String> {
    let mut txn = doc.transact_mut_with(BOOTSTRAP_ORIGIN);
    let cell_formats = txn.get_or_insert_map(CELL_FORMATS);
    sync_cell_formats(&cell_formats, &mut txn, &model.styles)?;
    let meta = txn.get_or_insert_map(META);
    meta.insert(&mut txn, BASE_FINGERPRINT, base.fingerprint.as_str());
    meta.insert(&mut txn, "schemaVersion", SCHEMA_VERSION);
    meta.insert(&mut txn, STRUCTURE_GENERATION, 0_i64);
    let order = txn.get_or_insert_array(SHEET_ORDER);
    order.insert_range(&mut txn, 0, keys.iter().cloned());
    let sheets = txn.get_or_insert_map(SHEETS);
    for (key, sheet) in keys.iter().zip(&model.sheets) {
        let sheet_map = sheets.insert(&mut txn, key.as_str(), MapPrelim::default());
        sync_sheet(&sheet_map, &mut txn, sheet, &model.styles)?;
    }
    Ok(())
}

fn hydrate_doc(doc: &Doc, update: &[u8]) -> Result<(), String> {
    let update = decode_update_v1(update)?;
    doc.transact_mut_with(HYDRATE_ORIGIN)
        .apply_update(update)
        .map_err(|error| error.to_string())
}

fn hydrate_local_doc(doc: &Doc, update: &[u8]) -> Result<(), String> {
    let update = decode_local_update_v1(update)?;
    doc.transact_mut_with(HYDRATE_ORIGIN)
        .apply_update(update)
        .map_err(|error| error.to_string())
}

fn has_pending(doc: &Doc) -> bool {
    let txn = doc.transact();
    txn.store().pending_update().is_some() || txn.store().pending_ds().is_some()
}

fn build_undo_manager(
    doc: &Doc,
    undo_stack: Vec<StackItem<()>>,
    redo_stack: Vec<StackItem<()>>,
) -> Result<UndoManager<()>, String> {
    let txn = doc.transact();
    let sheets = txn
        .get_map(SHEETS)
        .ok_or_else(|| "missing sheet map".to_string())?;
    drop(txn);
    let options = UndoOptions {
        capture_timeout_millis: UNDO_CAPTURE_TIMEOUT_MS,
        tracked_origins: HashSet::from([Origin::from(doc.client_id().get())]),
        capture_transaction: None,
        timestamp: undo_clock(),
        init_undo_stack: undo_stack,
        init_redo_stack: redo_stack,
    };
    let mut undo = UndoManager::with_options(options);
    undo.expand_scope(doc, &sheets);
    Ok(undo)
}

fn undo_clock() -> Arc<dyn Clock> {
    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
    {
        Arc::new(|| {
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|elapsed| elapsed.as_millis() as u64)
                .unwrap_or_default()
        })
    }
    #[cfg(all(target_family = "wasm", target_os = "unknown"))]
    {
        use std::sync::atomic::{AtomicU64, Ordering};
        let ticks = AtomicU64::new(0);
        Arc::new(move || ticks.fetch_add(UNDO_CAPTURE_TIMEOUT_MS + 1, Ordering::Relaxed))
    }
}

fn decode_state_vector_v1(bytes: &[u8]) -> Result<StateVector, String> {
    validate_state_vector_entry_count(bytes)?;
    let mut decoder = CheckedDecoderV1::new(bytes);
    let entries = decoder
        .read_var_u32_checked()
        .map_err(|error| error.to_string())?;
    let mut state_vector = StateVector::default();
    for _ in 0..entries {
        let client = decoder
            .read_var_u64_checked()
            .map_err(|error| error.to_string())?;
        let client = checked_client_id(client).map_err(|error| error.to_string())?;
        if state_vector.contains_client(&client) {
            return Err("state vector contains a duplicate client".to_string());
        }
        let clock = decoder
            .read_var_u32_checked()
            .map_err(|error| error.to_string())?;
        checked_clock(clock).map_err(|error| error.to_string())?;
        state_vector.set_max(client, clock);
    }
    if !decoder
        .read_to_end()
        .map_err(|error| error.to_string())?
        .is_empty()
    {
        return Err("state vector contains trailing bytes".to_string());
    }
    Ok(state_vector)
}

fn decode_update_v1(bytes: &[u8]) -> Result<Update, String> {
    decode_checked_update(CheckedDecoderV1::new_update(bytes))
}

/// Decodes authority-generated bytes with counts bounded by their encoded size.
fn decode_local_update_v1(bytes: &[u8]) -> Result<Update, String> {
    let mut decoder = CheckedDecoderV1::new_update(bytes);
    decoder.max_blocks = bytes.len();
    decoder.max_values = bytes.len();
    decoder.max_delete_ranges = bytes.len();
    decode_checked_update(decoder)
}

fn decode_checked_update(mut decoder: CheckedDecoderV1<'_>) -> Result<Update, String> {
    catch_unwind(AssertUnwindSafe(|| {
        let update = Update::decode(&mut decoder).map_err(|error| error.to_string())?;
        if !decoder
            .read_to_end()
            .map_err(|error| error.to_string())?
            .is_empty()
        {
            return Err("update contains trailing bytes".to_string());
        }
        Ok(update)
    }))
    .map_err(|_| "update decoder panicked".to_string())?
}

#[derive(Clone, Copy)]
enum CaptureKind {
    BlockClock,
    BlockCount,
    ClientCount,
    DeleteClient,
    DeleteClientCount,
    DeleteRangeCount,
    SkipLength,
}

struct VarCapture {
    kind: CaptureKind,
    bytes: u8,
    shift: u32,
    value: u64,
}

impl VarCapture {
    fn new(kind: CaptureKind) -> Self {
        Self {
            kind,
            bytes: 0,
            shift: 0,
            value: 0,
        }
    }
}

#[derive(Clone, Copy)]
enum PendingLength {
    Any,
    Deleted,
    Gc,
    Type,
}

struct CheckedDecoderV1<'a> {
    inner: DecoderV1<'a>,
    remaining: usize,
    max_blocks: usize,
    max_values: usize,
    max_delete_ranges: usize,
    clock: Option<u32>,
    delete_clock: Option<u32>,
    capture: Option<VarCapture>,
    pending_length: Option<PendingLength>,
    update_mode: bool,
    update_clients_remaining: u32,
    blocks_remaining: u32,
    total_blocks: usize,
    any_items_remaining: u32,
    any_block_len: u32,
    declared_any_items: usize,
    decoded_any_values: usize,
    delete_clients_remaining: u32,
    delete_ranges_remaining: u32,
    total_delete_ranges: usize,
}

impl<'a> CheckedDecoderV1<'a> {
    fn new(bytes: &'a [u8]) -> Self {
        Self {
            inner: DecoderV1::from(bytes),
            remaining: bytes.len(),
            max_blocks: MAX_UPDATE_BLOCKS,
            max_values: MAX_UPDATE_VALUES,
            max_delete_ranges: MAX_UPDATE_DELETE_RANGES,
            clock: None,
            delete_clock: None,
            capture: None,
            pending_length: None,
            update_mode: false,
            update_clients_remaining: 0,
            blocks_remaining: 0,
            total_blocks: 0,
            any_items_remaining: 0,
            any_block_len: 0,
            declared_any_items: 0,
            decoded_any_values: 0,
            delete_clients_remaining: 0,
            delete_ranges_remaining: 0,
            total_delete_ranges: 0,
        }
    }

    fn new_update(bytes: &'a [u8]) -> Self {
        let mut decoder = Self::new(bytes);
        decoder.update_mode = true;
        decoder.capture = Some(VarCapture::new(CaptureKind::ClientCount));
        decoder
    }

    fn begin_capture(&mut self, kind: CaptureKind) -> Result<(), DecodeError> {
        if self.capture.is_some() {
            return Err(decode_error("overlapping update varints"));
        }
        self.capture = Some(VarCapture::new(kind));
        Ok(())
    }

    fn feed_capture(&mut self, byte: u8) -> Result<(), DecodeError> {
        let Some(mut capture) = self.capture.take() else {
            return Ok(());
        };
        capture.bytes = capture
            .bytes
            .checked_add(1)
            .ok_or(DecodeError::InvalidVarInt)?;
        let payload = u64::from(byte & 0x7f);
        if capture.shift >= 64 || payload > (u64::MAX >> capture.shift) {
            return Err(DecodeError::InvalidVarInt);
        }
        capture.value |= payload << capture.shift;
        if byte & 0x80 != 0 {
            if capture.shift >= 63 {
                return Err(DecodeError::InvalidVarInt);
            }
            capture.shift += 7;
            self.capture = Some(capture);
            return Ok(());
        }
        if capture.bytes > 1 && payload == 0 {
            return Err(DecodeError::InvalidVarInt);
        }

        match capture.kind {
            CaptureKind::BlockClock => {
                let clock = u32::try_from(capture.value).map_err(|_| DecodeError::InvalidVarInt)?;
                checked_clock(clock)?;
                self.clock = Some(clock);
            }
            CaptureKind::BlockCount => {
                let blocks =
                    u32::try_from(capture.value).map_err(|_| DecodeError::InvalidVarInt)?;
                if self.update_clients_remaining == 0 || blocks == 0 {
                    return Err(decode_error("update contains an empty client block set"));
                }
                let blocks = blocks as usize;
                self.total_blocks = self
                    .total_blocks
                    .checked_add(blocks)
                    .ok_or_else(|| decode_error("update block count overflow"))?;
                if self.total_blocks > self.max_blocks || blocks > self.remaining {
                    return Err(decode_error("update contains too many blocks"));
                }
                self.blocks_remaining = blocks as u32;
            }
            CaptureKind::ClientCount => {
                let clients =
                    u32::try_from(capture.value).map_err(|_| DecodeError::InvalidVarInt)?;
                if clients > MAX_STATE_VECTOR_ENTRIES {
                    return Err(decode_error("update contains too many clients"));
                }
                self.update_clients_remaining = clients;
                self.begin_capture(if clients == 0 {
                    CaptureKind::DeleteClientCount
                } else {
                    CaptureKind::BlockCount
                })?;
            }
            CaptureKind::DeleteClient => {
                checked_client_id(capture.value)?;
                self.begin_capture(CaptureKind::DeleteRangeCount)?;
            }
            CaptureKind::DeleteClientCount => {
                let clients =
                    u32::try_from(capture.value).map_err(|_| DecodeError::InvalidVarInt)?;
                if clients > MAX_STATE_VECTOR_ENTRIES {
                    return Err(decode_error("update delete set contains too many clients"));
                }
                self.delete_clients_remaining = clients;
            }
            CaptureKind::DeleteRangeCount => {
                let ranges =
                    u32::try_from(capture.value).map_err(|_| DecodeError::InvalidVarInt)?;
                let total = self
                    .total_delete_ranges
                    .checked_add(ranges as usize)
                    .ok_or_else(|| decode_error("update delete range count overflow"))?;
                if total > self.max_delete_ranges || ranges as usize > self.remaining {
                    return Err(decode_error("update contains too many delete ranges"));
                }
                self.total_delete_ranges = total;
                self.delete_ranges_remaining = ranges;
                if ranges == 0 {
                    self.finish_delete_client()?;
                }
            }
            CaptureKind::SkipLength => {
                let len = u32::try_from(capture.value).map_err(|_| DecodeError::InvalidVarInt)?;
                if len == 0 {
                    return Err(decode_error("update contains an empty skip block"));
                }
                self.advance_clock(len)?;
                self.finish_block()?;
            }
        }
        Ok(())
    }

    fn advance_clock(&mut self, len: u32) -> Result<(), DecodeError> {
        let clock = self
            .clock
            .ok_or_else(|| decode_error("update block is missing its initial clock"))?;
        let next = clock
            .checked_add(len)
            .ok_or_else(|| decode_error("update block clock overflows u32"))?;
        checked_clock(next)?;
        self.clock = Some(next);
        Ok(())
    }

    fn finish_block(&mut self) -> Result<(), DecodeError> {
        if self.blocks_remaining == 0 {
            return Err(decode_error("update contains more blocks than declared"));
        }
        self.blocks_remaining -= 1;
        if self.blocks_remaining == 0 {
            self.clock = None;
            self.update_clients_remaining = self
                .update_clients_remaining
                .checked_sub(1)
                .ok_or_else(|| decode_error("update client count underflow"))?;
            self.begin_capture(if self.update_clients_remaining == 0 {
                CaptureKind::DeleteClientCount
            } else {
                CaptureKind::BlockCount
            })?;
        }
        Ok(())
    }

    fn finish_delete_client(&mut self) -> Result<(), DecodeError> {
        self.delete_clients_remaining = self
            .delete_clients_remaining
            .checked_sub(1)
            .ok_or_else(|| decode_error("update delete client count underflow"))?;
        Ok(())
    }

    fn read_var_u64_checked(&mut self) -> Result<u64, DecodeError> {
        if self.capture.is_some() {
            return Err(decode_error(
                "unexpected checked varint during update framing",
            ));
        }
        let mut value = 0_u64;
        let mut shift = 0_u32;
        let mut bytes = 0_u8;
        loop {
            let byte = self.read_u8()?;
            bytes += 1;
            let payload = u64::from(byte & 0x7f);
            if shift >= 64 || payload > (u64::MAX >> shift) {
                return Err(DecodeError::InvalidVarInt);
            }
            value |= payload << shift;
            if byte & 0x80 == 0 {
                if bytes > 1 && payload == 0 {
                    return Err(DecodeError::InvalidVarInt);
                }
                return Ok(value);
            }
            if shift >= 63 {
                return Err(DecodeError::InvalidVarInt);
            }
            shift += 7;
        }
    }

    fn read_var_u32_checked(&mut self) -> Result<u32, DecodeError> {
        u32::try_from(self.read_var_u64_checked()?).map_err(|_| DecodeError::InvalidVarInt)
    }

    fn read_var_usize_checked(&mut self) -> Result<usize, DecodeError> {
        usize::try_from(self.read_var_u64_checked()?).map_err(|_| DecodeError::InvalidVarInt)
    }

    fn read_var_i64_checked(&mut self) -> Result<i64, DecodeError> {
        let first = self.read_u8()?;
        let negative = first & 0x40 != 0;
        let mut magnitude = u64::from(first & 0x3f);
        let mut byte = first;
        let mut shift = 6_u32;
        while byte & 0x80 != 0 {
            byte = self.read_u8()?;
            let payload = u64::from(byte & 0x7f);
            if shift >= 63 || payload > (i64::MAX as u64 >> shift) {
                return Err(DecodeError::InvalidVarInt);
            }
            magnitude |= payload << shift;
            if byte & 0x80 == 0 {
                if payload == 0 {
                    return Err(DecodeError::InvalidVarInt);
                }
                break;
            }
            shift += 7;
        }
        if negative && magnitude == 0 {
            return Err(DecodeError::InvalidVarInt);
        }
        let value = i64::try_from(magnitude).map_err(|_| DecodeError::InvalidVarInt)?;
        Ok(if negative { -value } else { value })
    }

    fn read_id(&mut self) -> Result<ID, DecodeError> {
        let client = checked_client_id(self.read_var_u64_checked()?)?;
        let clock = self.read_var_u32_checked()?;
        checked_clock(clock)?;
        Ok(ID::new(client, clock))
    }

    fn decode_any(&mut self, depth: u8) -> Result<Any, DecodeError> {
        if depth >= 64 {
            return Err(decode_error("update value nesting exceeds 64 levels"));
        }
        self.decoded_any_values = self
            .decoded_any_values
            .checked_add(1)
            .ok_or_else(|| decode_error("update value count overflow"))?;
        if self.decoded_any_values > self.max_values {
            return Err(decode_error("update contains too many values"));
        }
        Ok(match self.read_u8()? {
            127 => Any::Undefined,
            126 => Any::Null,
            125 => Any::Number(self.read_var_i64_checked()? as f64),
            124 => Any::Number(self.read_f32()? as f64),
            123 => Any::Number(self.read_f64()?),
            122 => Any::BigInt(self.read_i64()?),
            121 => Any::Bool(false),
            120 => Any::Bool(true),
            119 => Any::String(Arc::from(self.read_string()?)),
            118 => {
                let len = self.read_var_usize_checked()?;
                if len > self.remaining
                    || len > self.max_values.saturating_sub(self.decoded_any_values)
                {
                    return Err(decode_error("update map length exceeds its payload"));
                }
                let mut map = HashMap::new();
                map.try_reserve(len)?;
                for _ in 0..len {
                    let key = self.read_string()?.to_owned();
                    map.insert(key, self.decode_any(depth + 1)?);
                }
                Any::Map(Arc::new(map))
            }
            117 => {
                let len = self.read_var_usize_checked()?;
                if len > self.remaining
                    || len > self.max_values.saturating_sub(self.decoded_any_values)
                {
                    return Err(decode_error("update array length exceeds its payload"));
                }
                let mut values = Vec::new();
                values.try_reserve(len)?;
                for _ in 0..len {
                    values.push(self.decode_any(depth + 1)?);
                }
                Any::Array(Arc::from(values))
            }
            116 => {
                let len = self.read_var_u32_checked()? as usize;
                Any::Buffer(Arc::from(self.read_exact(len)?))
            }
            _ => return Err(DecodeError::UnexpectedValue),
        })
    }
}

impl Read for CheckedDecoderV1<'_> {
    fn read_exact(&mut self, len: usize) -> Result<&[u8], DecodeError> {
        if len > self.remaining {
            return Err(DecodeError::EndOfBuffer(len));
        }
        let bytes = self.inner.read_exact(len)?;
        self.remaining -= len;
        Ok(bytes)
    }

    fn read_u8(&mut self) -> Result<u8, DecodeError> {
        if self.remaining == 0 {
            return Err(DecodeError::EndOfBuffer(1));
        }
        let byte = self.inner.read_u8()?;
        self.remaining -= 1;
        self.feed_capture(byte)?;
        Ok(byte)
    }

    fn read_string(&mut self) -> Result<&str, DecodeError> {
        let len = self.read_var_u32_checked()? as usize;
        let bytes = self.read_exact(len)?;
        std::str::from_utf8(bytes).map_err(|error| decode_error(error.to_string()))
    }
}

impl Decoder for CheckedDecoderV1<'_> {
    fn reset_ds_cur_val(&mut self) {
        self.delete_clock = None;
        if self.capture.is_none()
            && self.delete_clients_remaining > 0
            && self.delete_ranges_remaining == 0
        {
            self.capture = Some(VarCapture::new(CaptureKind::DeleteClient));
        }
    }

    fn read_ds_clock(&mut self) -> Result<u32, DecodeError> {
        if self.delete_ranges_remaining == 0 {
            return Err(decode_error("update contains an undeclared delete range"));
        }
        let clock = self.read_var_u32_checked()?;
        checked_clock(clock)?;
        self.delete_clock = Some(clock);
        Ok(clock)
    }

    fn read_ds_len(&mut self) -> Result<u32, DecodeError> {
        let len = self.read_var_u32_checked()?;
        if len == 0 {
            return Err(decode_error("update contains an empty delete range"));
        }
        let end = self
            .delete_clock
            .ok_or_else(|| decode_error("delete range is missing its clock"))?
            .checked_add(len)
            .ok_or_else(|| decode_error("delete range clock overflows u32"))?;
        checked_clock(end)?;
        self.delete_ranges_remaining -= 1;
        if self.delete_ranges_remaining == 0 {
            self.finish_delete_client()?;
        }
        Ok(len)
    }

    fn read_left_id(&mut self) -> Result<ID, DecodeError> {
        self.read_id()
    }

    fn read_right_id(&mut self) -> Result<ID, DecodeError> {
        self.read_id()
    }

    fn read_client(&mut self) -> Result<ClientID, DecodeError> {
        if self.blocks_remaining == 0 {
            return Err(decode_error("update client is missing its block count"));
        }
        let client = checked_client_id(self.read_var_u64_checked()?)?;
        self.begin_capture(CaptureKind::BlockClock)?;
        Ok(client)
    }

    fn read_info(&mut self) -> Result<u8, DecodeError> {
        if self.pending_length.is_some() {
            return Err(decode_error("update block is missing its content length"));
        }
        let info = self.read_u8()?;
        if info == BLOCK_SKIP_REF_NUMBER {
            self.begin_capture(CaptureKind::SkipLength)?;
        } else if info == BLOCK_GC_REF_NUMBER {
            self.pending_length = Some(PendingLength::Gc);
        } else {
            match info & 0x0f {
                BLOCK_ITEM_DELETED_REF_NUMBER => {
                    self.pending_length = Some(PendingLength::Deleted);
                }
                BLOCK_ITEM_TYPE_REF_NUMBER => {
                    self.pending_length = Some(PendingLength::Type);
                }
                BLOCK_ITEM_ANY_REF_NUMBER => {
                    self.pending_length = Some(PendingLength::Any);
                }
                _ => return Err(decode_error("update contains an unsupported block type")),
            }
        }
        Ok(info)
    }

    fn read_parent_info(&mut self) -> Result<bool, DecodeError> {
        match self.read_var_u32_checked()? {
            0 => Ok(false),
            1 => Ok(true),
            _ => Err(decode_error("update contains invalid parent info")),
        }
    }

    fn read_type_ref(&mut self) -> Result<u8, DecodeError> {
        if !matches!(self.pending_length.take(), Some(PendingLength::Type)) {
            return Err(decode_error("update contains an unexpected shared type"));
        }
        let type_ref = self.read_u8()?;
        if type_ref != TYPE_REFS_ARRAY && type_ref != TYPE_REFS_MAP {
            return Err(decode_error("update contains an unsupported shared type"));
        }
        self.advance_clock(1)?;
        self.finish_block()?;
        Ok(type_ref)
    }

    fn read_len(&mut self) -> Result<u32, DecodeError> {
        let len = self.read_var_u32_checked()?;
        let pending = self
            .pending_length
            .take()
            .ok_or_else(|| decode_error("update contains an unexpected block length"))?;
        match pending {
            PendingLength::Gc => {
                if len == 0 {
                    return Err(decode_error("update contains an empty GC block"));
                }
                self.advance_clock(len)?;
                self.finish_block()?;
            }
            PendingLength::Deleted => {
                self.advance_clock(len)?;
                self.finish_block()?;
            }
            PendingLength::Any => {
                self.declared_any_items = self
                    .declared_any_items
                    .checked_add(len as usize)
                    .ok_or_else(|| decode_error("update value count overflow"))?;
                if self.declared_any_items > self.max_values || len as usize > self.remaining {
                    return Err(decode_error("update contains too many values"));
                }
                self.any_items_remaining = len;
                self.any_block_len = len;
                if len == 0 {
                    self.finish_block()?;
                }
            }
            PendingLength::Type => {
                return Err(decode_error("shared type block contains a length"));
            }
        }
        Ok(len)
    }

    fn read_any(&mut self) -> Result<Any, DecodeError> {
        if self.any_items_remaining == 0 {
            return Err(decode_error("update contains an undeclared value"));
        }
        let value = self.decode_any(0)?;
        self.any_items_remaining -= 1;
        if self.any_items_remaining == 0 {
            self.advance_clock(self.any_block_len)?;
            self.any_block_len = 0;
            self.finish_block()?;
        }
        Ok(value)
    }

    fn read_json(&mut self) -> Result<Any, DecodeError> {
        Any::from_json(self.read_string()?)
    }

    fn read_key(&mut self) -> Result<Arc<str>, DecodeError> {
        Ok(Arc::from(self.read_string()?))
    }

    fn read_to_end(&mut self) -> Result<&[u8], DecodeError> {
        if self.update_mode
            && (self.capture.is_some()
                || self.update_clients_remaining != 0
                || self.blocks_remaining != 0
                || self.pending_length.is_some()
                || self.any_items_remaining != 0
                || self.delete_clients_remaining != 0
                || self.delete_ranges_remaining != 0)
        {
            return Err(decode_error("update ended before its declared content"));
        }
        let bytes = self.inner.read_to_end()?;
        if bytes.len() != self.remaining {
            return Err(decode_error("update decoder length mismatch"));
        }
        Ok(bytes)
    }
}

fn checked_client_id(client: u64) -> Result<ClientID, DecodeError> {
    if client > MAX_SAFE_CLIENT_ID {
        Err(decode_error("client ID exceeds the 53-bit Yjs limit"))
    } else {
        Ok(ClientID::new(client))
    }
}

fn checked_clock(clock: u32) -> Result<(), DecodeError> {
    if clock > MAX_SAFE_CLOCK {
        Err(decode_error("clock exceeds the supported i32 range"))
    } else {
        Ok(())
    }
}

fn decode_error(message: impl Into<String>) -> DecodeError {
    DecodeError::Custom(message.into())
}

fn validate_state_vector_entry_count(bytes: &[u8]) -> Result<(), String> {
    let Some((&first, _)) = bytes.split_first() else {
        return Err("state vector is empty".to_string());
    };
    let mut value = u32::from(first & 0x7f);
    let mut shift = 7;
    let mut used = 1;
    let mut byte = first;
    while byte & 0x80 != 0 {
        if used == 5 || used >= bytes.len() {
            return Err("invalid state vector entry count".to_string());
        }
        byte = bytes[used];
        if used == 4 && byte > 0x0f {
            return Err("invalid state vector entry count".to_string());
        }
        value |= u32::from(byte & 0x7f) << shift;
        shift += 7;
        used += 1;
    }
    if value > MAX_STATE_VECTOR_ENTRIES {
        return Err(format!(
            "state vector contains {value} entries, exceeds the {MAX_STATE_VECTOR_ENTRIES}-entry limit"
        ));
    }
    if value as usize > bytes.len().saturating_sub(used) / 2 {
        return Err("state vector entry count exceeds its payload".to_string());
    }
    Ok(())
}

fn requires_full_semantic_sync(op: &Op) -> bool {
    matches!(
        op,
        Op::InsertRows { .. }
            | Op::DeleteRows { .. }
            | Op::InsertCols { .. }
            | Op::DeleteCols { .. }
            | Op::SetFreezePane { .. }
            | Op::SetHyperlinks { .. }
            | Op::RestoreColStyles { .. }
            | Op::SetCharts { .. }
            | Op::SetChartAnchor { .. }
            | Op::RemoveSheet { .. }
            | Op::RenameSheet { .. }
            | Op::RestoreSheet { .. }
    )
}

#[derive(Clone)]
enum SheetToken {
    Existing(String),
    Added(usize),
}

fn targeted_sheet_keys(
    current: &[String],
    desired: &[String],
    ops: &[Op],
) -> Result<Vec<Option<String>>, String> {
    let mut tokens = current
        .iter()
        .cloned()
        .map(SheetToken::Existing)
        .collect::<Vec<_>>();
    let mut targets = Vec::with_capacity(ops.len());
    let mut next_added = 0;
    for op in ops {
        match op {
            Op::AddSheet { index, .. } => {
                if *index > tokens.len() {
                    return Err(format!("sheet insertion index {index} is out of range"));
                }
                tokens.insert(*index, SheetToken::Added(next_added));
                next_added += 1;
                targets.push(None);
            }
            Op::RemoveSheet { index } => {
                if *index >= tokens.len() {
                    return Err(format!("sheet removal index {index} is out of range"));
                }
                tokens.remove(*index);
                targets.push(None);
            }
            Op::SetDefinedNames { .. } => targets.push(None),
            op => {
                let sheet = op_sheet(op)
                    .ok_or_else(|| "operation has no sheet target".to_string())?
                    .0 as usize;
                let token = tokens
                    .get(sheet)
                    .cloned()
                    .ok_or_else(|| format!("sheet {sheet} is out of range"))?;
                targets.push(Some(token));
            }
        }
    }
    if tokens.len() != desired.len() {
        return Err("sheet operation plan does not match final order".to_string());
    }
    let mut added = HashMap::new();
    for (token, key) in tokens.iter().zip(desired) {
        match token {
            SheetToken::Existing(existing) if existing != key => {
                return Err("sheet operation plan changed retained identity".to_string());
            }
            SheetToken::Existing(_) => {}
            SheetToken::Added(id) => {
                added.insert(*id, key.clone());
            }
        }
    }
    let active = desired.iter().map(String::as_str).collect::<HashSet<_>>();
    Ok(targets
        .into_iter()
        .map(|target| match target {
            Some(SheetToken::Existing(key)) if active.contains(key.as_str()) => Some(key),
            Some(SheetToken::Added(id)) => added.get(&id).cloned(),
            _ => None,
        })
        .collect())
}

fn op_sheet(op: &Op) -> Option<SheetId> {
    match op {
        Op::SetCell { sheet, .. }
        | Op::InsertRows { sheet, .. }
        | Op::DeleteRows { sheet, .. }
        | Op::InsertCols { sheet, .. }
        | Op::DeleteCols { sheet, .. }
        | Op::SetColWidth { sheet, .. }
        | Op::SetRowHeight { sheet, .. }
        | Op::SetFreezePane { sheet, .. }
        | Op::SetHyperlinks { sheet, .. }
        | Op::RestoreColStyles { sheet, .. }
        | Op::SetCharts { sheet, .. }
        | Op::SetChartAnchor { sheet, .. }
        | Op::MergeCells { sheet, .. }
        | Op::UnmergeCells { sheet, .. }
        | Op::PatchRangeStyle { sheet, .. }
        | Op::SetRangeNumberFormat { sheet, .. }
        | Op::ApplyRangeFormat { sheet, .. }
        | Op::RenameSheet { sheet, .. }
        | Op::RestoreSheet { sheet, .. } => Some(*sheet),
        Op::AddSheet { .. } | Op::RemoveSheet { .. } | Op::SetDefinedNames { .. } => None,
    }
}

fn patch_sheet_order(
    order: &ArrayRef,
    txn: &mut TransactionMut<'_>,
    existing: &[String],
    desired: &[String],
) -> Result<(), String> {
    let mut working = existing.to_vec();
    let mut index = 0;
    while index < desired.len() {
        if working.get(index) == desired.get(index) {
            index += 1;
            continue;
        }
        if let Some(offset) = working[index..]
            .iter()
            .position(|key| key == &desired[index])
        {
            order.remove_range(txn, yrs_index(index)?, yrs_index(offset)?);
            working.drain(index..index + offset);
        } else {
            order.insert(txn, yrs_index(index)?, desired[index].clone());
            working.insert(index, desired[index].clone());
            index += 1;
        }
    }
    if working.len() > desired.len() {
        order.remove_range(
            txn,
            yrs_index(desired.len())?,
            yrs_index(working.len() - desired.len())?,
        );
    }
    Ok(())
}

fn yrs_index(index: usize) -> Result<u32, String> {
    u32::try_from(index).map_err(|_| "sheet order exceeds Yrs index range".to_string())
}

fn sheet_map_for_sync(
    sheets: &MapRef,
    txn: &mut TransactionMut<'_>,
    key: &str,
) -> Result<MapRef, String> {
    match sheets.get(txn, key) {
        Some(Out::YMap(map)) => Ok(map),
        Some(_) => Err(format!("sheet {key} is not a map")),
        None => Ok(sheets.insert(txn, key, MapPrelim::default())),
    }
}

fn sheet_parts_by_key<'a, T: ReadTxn>(
    sheets: &MapRef,
    txn: &T,
    keys: &[String],
    model: &'a WorkbookModel,
    key: &str,
) -> Result<(MapRef, &'a Sheet), String> {
    let index = keys
        .iter()
        .position(|candidate| candidate == key)
        .ok_or_else(|| format!("sheet {key} is not active"))?;
    let sheet_map = sheets
        .get(txn, key)
        .and_then(|value| value.cast::<MapRef>().ok())
        .ok_or_else(|| format!("missing sheet {key}"))?;
    let sheet_model = model
        .sheets
        .get(index)
        .ok_or_else(|| format!("sheet {key} is missing from the projection"))?;
    Ok((sheet_map, sheet_model))
}

fn sync_sheet(
    sheet_map: &MapRef,
    txn: &mut TransactionMut<'_>,
    sheet: &Sheet,
    stylesheet: &Stylesheet,
) -> Result<(), String> {
    let col_widths: MapRef = sheet_map.get_or_init(txn, COL_WIDTHS);
    let contents: MapRef = sheet_map.get_or_init(txn, CONTENTS);
    sheet_map.try_update(txn, FREEZE_PANE, freeze_pane_to_any(sheet.freeze_pane));
    let hyperlinks = serde_json::to_string(&sheet.hyperlinks)
        .map_err(|error| format!("cannot encode sheet hyperlinks: {error}"))?;
    sheet_map.try_update(txn, HYPERLINKS, hyperlinks);
    let charts = serde_json::to_string(&sheet.charts)
        .map_err(|error| format!("cannot encode sheet charts: {error}"))?;
    sheet_map.try_update(txn, CHARTS, charts);
    sheet_map.try_update(txn, MERGES, merges_to_any(&sheet.merges));
    sheet_map.try_update(txn, NAME, sheet.name.as_str());
    let row_heights: MapRef = sheet_map.get_or_init(txn, ROW_HEIGHTS);
    let styles: MapRef = sheet_map.get_or_init(txn, STYLES);
    sync_numbers(&col_widths, txn, &sheet.col_widths);
    sync_contents(&contents, txn, sheet);
    sync_numbers(&row_heights, txn, &sheet.row_heights);
    sync_styles(&styles, txn, sheet, stylesheet)?;
    Ok(())
}

fn sync_contents(map: &MapRef, txn: &mut TransactionMut<'_>, sheet: &Sheet) {
    let desired = sheet
        .iter_cells()
        .filter_map(|(at, cell)| content_to_any(cell).map(|value| (cell_key(at), value)))
        .collect::<BTreeMap<_, _>>();
    sync_map(map, txn, desired);
}

fn sync_styles(
    map: &MapRef,
    txn: &mut TransactionMut<'_>,
    sheet: &Sheet,
    stylesheet: &Stylesheet,
) -> Result<(), String> {
    let desired = sheet
        .iter_cells()
        .filter_map(|(at, cell)| cell.style.map(|style| (at, style)))
        .map(|(at, style)| style_key(stylesheet, style).map(|key| (cell_key(at), Any::from(key))))
        .collect::<Result<BTreeMap<_, _>, _>>()?;
    sync_map(map, txn, desired);
    Ok(())
}

fn sync_map(map: &MapRef, txn: &mut TransactionMut<'_>, desired: BTreeMap<String, Any>) {
    let mut stale = map
        .keys(txn)
        .filter(|key| !desired.contains_key(*key))
        .map(str::to_string)
        .collect::<Vec<_>>();
    stale.sort();
    for key in stale {
        map.remove(txn, &key);
    }
    for (key, value) in desired {
        map.try_update(txn, key, value);
    }
}

fn sync_authored_cell(
    sheet_map: &MapRef,
    txn: &mut TransactionMut<'_>,
    sheet: &Sheet,
    at: CellRef,
) -> Result<(), String> {
    let contents: MapRef = sheet_map.get_or_init(txn, CONTENTS);
    let key = cell_key(at);
    match sheet.cell(at) {
        Some(cell) => {
            let current = match contents.get(txn, &key) {
                Some(Out::Any(value)) => Some(content_from_any(&value)?),
                Some(_) => return Err(format!("cell content {key} is not an atomic value")),
                None => None,
            };
            if !authored_content_equal(current.as_ref(), Some(cell)) {
                sync_optional(&contents, txn, &key, content_to_any(cell));
            }
        }
        None => {
            contents.remove(txn, &key);
        }
    }
    Ok(())
}

fn sync_cell_format(
    sheet_map: &MapRef,
    txn: &mut TransactionMut<'_>,
    sheet: &Sheet,
    stylesheet: &Stylesheet,
    at: CellRef,
) -> Result<(), String> {
    let styles: MapRef = sheet_map.get_or_init(txn, STYLES);
    let key = cell_key(at);
    let style = sheet
        .cell(at)
        .and_then(|cell| cell.style)
        .map(|style| style_key(stylesheet, style).map(Any::from))
        .transpose()?;
    sync_optional(&styles, txn, &key, style);
    Ok(())
}

fn sync_cell_formats(
    map: &MapRef,
    txn: &mut TransactionMut<'_>,
    stylesheet: &Stylesheet,
) -> Result<(), String> {
    let (key, payload) = cell_format_entry(&CellFormat::default())?;
    map.try_update(txn, key, payload);
    for index in 0..stylesheet.cell_xfs.len() {
        let index =
            u32::try_from(index).map_err(|_| "cell format table is too large".to_string())?;
        let format = stylesheet.cell_format(Some(index));
        let (key, payload) = cell_format_entry(&format)?;
        map.try_update(txn, key, payload);
    }
    Ok(())
}

fn materialize_cell_formats<T: ReadTxn>(
    map: &MapRef,
    txn: &T,
    base: &Stylesheet,
) -> Result<(Stylesheet, BTreeMap<String, Option<u32>>), String> {
    let mut catalog = BTreeMap::new();
    for (key, value) in map.iter(txn) {
        let Out::Any(Any::String(payload)) = value else {
            return Err(format!("cell format {key} is not a string"));
        };
        if payload.len() > MAX_CELL_FORMAT_BYTES {
            return Err(format!("cell format {key} exceeds its size limit"));
        }
        let format = serde_json::from_str::<CellFormat>(&payload)
            .map_err(|error| format!("invalid cell format {key}: {error}"))?;
        let (expected, canonical) = cell_format_entry(&format)?;
        if key != expected || *payload != canonical {
            return Err(format!("cell format {key} is not canonical"));
        }
        catalog.insert(key.to_string(), format);
    }

    let mut styles = base.clone();
    let mut known = BTreeMap::new();
    for index in 0..base.cell_xfs.len() {
        let index =
            u32::try_from(index).map_err(|_| "cell format table is too large".to_string())?;
        known.entry(style_key(base, index)?).or_insert(Some(index));
    }
    let mut indices = BTreeMap::new();
    for (key, format) in catalog {
        let index = match known.get(&key) {
            Some(index) => *index,
            None => {
                let index = styles
                    .intern_cell_format(&format)
                    .map_err(|_| "number format table is full".to_string())?;
                known.insert(key.clone(), index);
                index
            }
        };
        indices.insert(key, index);
    }
    Ok((styles, indices))
}

/// Rewrites the style indices `ops` carry from `source`'s table to the indices of the same
/// formats in `target`, adding formats `target` lacks. Two tables built from one base agree on
/// its indices but intern later formats in their own order.
fn remap_styles<'a>(
    ops: &'a [Op],
    source: &Stylesheet,
    target: &mut Stylesheet,
) -> Result<Cow<'a, [Op]>, String> {
    let mut mapped: HashMap<u32, Option<u32>> = HashMap::new();
    let mut resolve = |style: Option<u32>| -> Result<Option<u32>, String> {
        let Some(index) = style else {
            return Ok(None);
        };
        if let Some(&known) = mapped.get(&index) {
            return Ok(known);
        }
        if source.xf(index).is_none() {
            return Err(format!("cell style index {index} is out of range"));
        }
        let format = source.cell_format(Some(index));
        let resolved = if target.xf(index).is_some() && target.cell_format(Some(index)) == format {
            Some(index)
        } else {
            match (0..target.cell_xfs.len() as u32)
                .find(|&candidate| target.cell_format(Some(candidate)) == format)
            {
                Some(found) => Some(found),
                None => target
                    .intern_cell_format(&format)
                    .map_err(|_| "number format table is full".to_string())?,
            }
        };
        mapped.insert(index, resolved);
        Ok(resolved)
    };
    let mut remapped: Option<Vec<Op>> = None;
    for (position, op) in ops.iter().enumerate() {
        let rewritten = match op {
            Op::SetCell { sheet, at, cell } => {
                let style = resolve(cell.style)?;
                (style != cell.style).then(|| Op::SetCell {
                    sheet: *sheet,
                    at: *at,
                    cell: CellState {
                        style,
                        ..cell.clone()
                    },
                })
            }
            Op::RestoreSheet {
                sheet,
                name,
                formulas,
            } => {
                let mut changed = false;
                let mut restored = Vec::with_capacity(formulas.len());
                for (formula_sheet, at, state) in formulas {
                    let style = resolve(state.style)?;
                    changed |= style != state.style;
                    restored.push((
                        *formula_sheet,
                        *at,
                        CellState {
                            style,
                            ..state.clone()
                        },
                    ));
                }
                changed.then(|| Op::RestoreSheet {
                    sheet: *sheet,
                    name: name.clone(),
                    formulas: restored,
                })
            }
            _ => None,
        };
        if let Some(rewritten) = rewritten {
            remapped
                .get_or_insert_with(|| ops[..position].to_vec())
                .push(rewritten);
        } else if let Some(remapped) = &mut remapped {
            remapped.push(op.clone());
        }
    }
    Ok(match remapped {
        Some(remapped) => Cow::Owned(remapped),
        None => Cow::Borrowed(ops),
    })
}

/// Whether `format` fits the size a shared cell format may take.
pub(crate) fn cell_format_fits(format: &CellFormat) -> bool {
    cell_format_entry(format).is_ok_and(|(_, payload)| payload.len() <= MAX_CELL_FORMAT_BYTES)
}

fn style_key(stylesheet: &Stylesheet, style: u32) -> Result<String, String> {
    if stylesheet.xf(style).is_none() {
        return Err(format!("cell style index {style} is out of range"));
    }
    cell_format_entry(&stylesheet.cell_format(Some(style))).map(|(key, _)| key)
}

fn cell_format_entry(format: &CellFormat) -> Result<(String, String), String> {
    let payload = serde_json::to_string(format)
        .map_err(|error| format!("cannot encode cell format: {error}"))?;
    let digest = Sha256::digest(payload.as_bytes());
    Ok((format!("{digest:x}"), payload))
}

fn authored_content_equal(left: Option<&Cell>, right: Option<&Cell>) -> bool {
    match (left, right) {
        (None, None) => true,
        (Some(left), Some(right)) => match (&left.formula, &right.formula) {
            (Some(left), Some(right)) => left == right,
            (None, None) => left.value == right.value,
            _ => false,
        },
        _ => false,
    }
}

fn sync_optional(map: &MapRef, txn: &mut TransactionMut<'_>, key: &str, value: Option<Any>) {
    if let Some(value) = value {
        map.try_update(txn, key, value);
    } else {
        map.remove(txn, key);
    }
}

fn sync_numbers(map: &MapRef, txn: &mut TransactionMut<'_>, values: &BTreeMap<u32, f64>) {
    let retained = values.keys().map(u32::to_string).collect::<HashSet<_>>();
    let mut stale = map
        .keys(txn)
        .filter(|key| !retained.contains(*key))
        .map(str::to_string)
        .collect::<Vec<_>>();
    stale.sort();
    for key in stale {
        map.remove(txn, &key);
    }
    for (&index, &value) in values {
        map.try_update(txn, index.to_string(), value);
    }
}

fn sync_number(map: &MapRef, txn: &mut TransactionMut<'_>, index: u32, value: Option<f64>) {
    let key = index.to_string();
    match value {
        Some(value) => {
            map.try_update(txn, key, value);
        }
        None => {
            map.remove(txn, &key);
        }
    }
}

/// What a sheet reads from the workbook it was parsed from rather than from
/// the shared document, because the document's schema cannot express it.
#[derive(Clone, Copy)]
struct SheetFallbacks<'a> {
    freeze_pane: Option<FreezePane>,
    format: SheetFormat,
    col_styles: &'a [ColStyle],
    hyperlinks: &'a [Hyperlink],
    charts: &'a [SheetChart],
    hidden_dimensions: &'a HiddenDimensions,
}

impl Default for SheetFallbacks<'_> {
    fn default() -> Self {
        Self {
            freeze_pane: None,
            format: SheetFormat::default(),
            col_styles: &[],
            hyperlinks: &[],
            charts: &[],
            hidden_dimensions: &EMPTY_HIDDEN_DIMENSIONS,
        }
    }
}

fn materialize_sheet<T: ReadTxn>(
    sheet_map: &MapRef,
    txn: &T,
    style_indices: &BTreeMap<String, Option<u32>>,
    version: i64,
    fallbacks: SheetFallbacks<'_>,
) -> Result<Sheet, String> {
    let name = sheet_map
        .get(txn, NAME)
        .and_then(|value| value.cast::<String>().ok())
        .ok_or_else(|| "sheet is missing its name".to_string())?;
    let mut sheet = Sheet::new(name);
    let mut cells = BTreeMap::<(u32, u32), Cell>::new();
    let contents = nested_map(sheet_map, txn, CONTENTS)?;
    for (key, value) in contents.iter(txn) {
        let at = parse_cell_key(key)?;
        let Out::Any(value) = value else {
            return Err(format!("cell content {key} is not an atomic value"));
        };
        cells.insert((at.row, at.col), content_from_any(&value)?);
    }
    let styles = nested_map(sheet_map, txn, STYLES)?;
    for (key, value) in styles.iter(txn) {
        let at = parse_cell_key(key)?;
        let style_key = value
            .cast::<String>()
            .map_err(|_| format!("cell style {key} is not a string"))?;
        let style = style_indices
            .get(&style_key)
            .ok_or_else(|| format!("cell style {key} references an unknown format"))?;
        cells.entry((at.row, at.col)).or_default().style = *style;
    }
    for ((row, col), cell) in cells {
        sheet.set_cell(CellRef::new(row, col), cell);
    }
    sheet.col_widths = materialize_numbers(
        &nested_map(sheet_map, txn, COL_WIDTHS)?,
        txn,
        MAX_COLS,
        "column width",
    )?;
    sheet.row_heights = materialize_numbers(
        &nested_map(sheet_map, txn, ROW_HEIGHTS)?,
        txn,
        MAX_ROWS,
        "row height",
    )?;
    if version < SCHEMA_VERSION {
        for (&at, &size) in &fallbacks.hidden_dimensions.col_widths {
            sheet.col_widths.entry(at).or_insert(size);
        }
        for (&at, &size) in &fallbacks.hidden_dimensions.row_heights {
            sheet.row_heights.entry(at).or_insert(size);
        }
    }
    sheet.format = fallbacks.format;
    sheet.col_styles = fallbacks.col_styles.to_vec();
    sheet.freeze_pane = match (version, sheet_map.get(txn, FREEZE_PANE)) {
        (FREEZE_PANE_SCHEMA_VERSION.., Some(Out::Any(value))) => freeze_pane_from_any(&value)?,
        (FREEZE_PANE_SCHEMA_VERSION.., _) => {
            return Err("sheet is missing freeze pane".to_string());
        }
        _ => fallbacks.freeze_pane,
    };
    sheet.hyperlinks = match (version, sheet_map.get(txn, HYPERLINKS)) {
        (HYPERLINK_SCHEMA_VERSION.., Some(Out::Any(Any::String(json)))) => {
            decode_hyperlinks(&json)?
        }
        (HYPERLINK_SCHEMA_VERSION.., Some(_)) => {
            return Err("sheet hyperlinks are not a string".to_string());
        }
        (HYPERLINK_SCHEMA_VERSION.., None) => {
            return Err("sheet is missing hyperlinks".to_string());
        }
        _ => fallbacks.hyperlinks.to_vec(),
    };
    sheet.charts = match (version, sheet_map.get(txn, CHARTS)) {
        (CHARTS_SCHEMA_VERSION.., Some(Out::Any(Any::String(json)))) => decode_charts(&json)?,
        (CHARTS_SCHEMA_VERSION.., Some(_)) => {
            return Err("sheet charts are not a string".to_string());
        }
        _ => fallbacks.charts.to_vec(),
    };
    sheet.merges = match sheet_map.get(txn, MERGES) {
        Some(Out::Any(value)) => merges_from_any(&value)?,
        _ => return Err("sheet is missing merges".to_string()),
    };
    Ok(sheet)
}

/// One drawing anchor is one element however many sheets point at it, but each
/// sheet's chart state is assigned on its own. Two replicas can each write a
/// legal value for a different sheet and only disagree once the two are merged,
/// so the disagreement is settled on the way out rather than refused on the way
/// in: refusing it would make integration depend on delivery order, and the
/// replicas would never meet again. Sheet order is shared, so first-in-order
/// wins is the same answer everywhere.
///
/// Removing a sheet is structural and refused while collaborative. Were that
/// ever allowed, dropping the donor would hand the frame to whatever the
/// surviving sheet's blob holds — which may be a value this has been quietly
/// covering for, and which nothing has checked since it arrived.
fn project_shared_frame_anchors(sheets: &mut [Sheet]) {
    let mut chosen = BTreeMap::new();
    for sheet in sheets.iter() {
        for chart in &sheet.charts {
            chosen.entry(chart.frame_id()).or_insert(chart.anchor);
        }
    }
    for sheet in sheets.iter_mut() {
        for chart in &mut sheet.charts {
            if let Some(anchor) = chosen.get(&chart.frame_id()) {
                chart.anchor = *anchor;
            }
        }
    }
}

fn nested_map<T: ReadTxn>(parent: &MapRef, txn: &T, key: &str) -> Result<MapRef, String> {
    parent
        .get(txn, key)
        .and_then(|value| value.cast::<MapRef>().ok())
        .ok_or_else(|| format!("sheet is missing {key}"))
}

fn sheet_shared_types<T: ReadTxn>(sheet_map: &MapRef, txn: &T) -> Result<SheetSharedTypes, String> {
    Ok(SheetSharedTypes {
        sheet: sheet_map.as_ref().id(),
        col_widths: nested_map(sheet_map, txn, COL_WIDTHS)?.as_ref().id(),
        contents: nested_map(sheet_map, txn, CONTENTS)?.as_ref().id(),
        row_heights: nested_map(sheet_map, txn, ROW_HEIGHTS)?.as_ref().id(),
        styles: nested_map(sheet_map, txn, STYLES)?.as_ref().id(),
    })
}

fn materialize_numbers<T: ReadTxn>(
    map: &MapRef,
    txn: &T,
    limit: u32,
    label: &str,
) -> Result<BTreeMap<u32, f64>, String> {
    let mut values = BTreeMap::new();
    for (key, value) in map.iter(txn) {
        let index = key
            .parse::<u32>()
            .map_err(|_| format!("invalid numeric key {key}"))?;
        if key != index.to_string() {
            return Err(format!("noncanonical numeric key {key}"));
        }
        if index >= limit {
            return Err(format!("{label} key {key} is out of bounds"));
        }
        let value = value
            .cast::<f64>()
            .map_err(|_| format!("invalid numeric value at {key}"))?;
        if !value.is_finite() {
            return Err(format!("nonfinite {label} at {key}"));
        }
        values.insert(index, value);
    }
    Ok(values)
}

fn sheet_keys<T: ReadTxn>(order: &ArrayRef, txn: &T) -> Result<Vec<String>, String> {
    order
        .iter(txn)
        .map(|value| {
            value
                .cast::<String>()
                .map_err(|_| "sheet order contains a non-string key".to_string())
        })
        .collect()
}

fn structure_generation<T: ReadTxn>(meta: &MapRef, txn: &T) -> Result<i64, String> {
    let generation = meta
        .get(txn, STRUCTURE_GENERATION)
        .and_then(|value| value.cast::<i64>().ok())
        .ok_or_else(|| "missing structure generation".to_string())?;
    if generation < 0 {
        return Err("structure generation is negative".to_string());
    }
    Ok(generation)
}

fn validate_schema_version(version: i64) -> Result<(), String> {
    if (MIN_SUPPORTED_SCHEMA_VERSION..=SCHEMA_VERSION).contains(&version) {
        Ok(())
    } else {
        Err(format!(
            "unsupported schema version {version}; supported versions are {MIN_SUPPORTED_SCHEMA_VERSION} through {SCHEMA_VERSION}"
        ))
    }
}

fn sheet_schema_keys(version: i64) -> &'static [&'static str] {
    const V3: &[&str] = &[COL_WIDTHS, CONTENTS, MERGES, NAME, ROW_HEIGHTS, STYLES];
    const V4: &[&str] = &[
        COL_WIDTHS,
        CONTENTS,
        FREEZE_PANE,
        MERGES,
        NAME,
        ROW_HEIGHTS,
        STYLES,
    ];
    const V5: &[&str] = &[
        COL_WIDTHS,
        CONTENTS,
        FREEZE_PANE,
        HYPERLINKS,
        MERGES,
        NAME,
        ROW_HEIGHTS,
        STYLES,
    ];
    const V6: &[&str] = &[
        COL_WIDTHS,
        CONTENTS,
        FREEZE_PANE,
        HYPERLINKS,
        MERGES,
        NAME,
        ROW_HEIGHTS,
        STYLES,
    ];
    match version {
        MIN_SUPPORTED_SCHEMA_VERSION => V3,
        FREEZE_PANE_SCHEMA_VERSION => V4,
        HYPERLINK_SCHEMA_VERSION => V5,
        _ => V6,
    }
}

/// Chart state is the one sheet key two replicas can assign at once, because
/// repinning a chart is an ordinary edit. Undoing the assignment that won
/// deletes the key outright, so it reads as absent and falls back to what the
/// source package holds rather than failing the whole workbook.
fn sheet_schema_optional_keys(version: i64) -> &'static [&'static str] {
    const NONE: &[&str] = &[];
    const V6: &[&str] = &[CHARTS];
    if version >= CHARTS_SCHEMA_VERSION {
        V6
    } else {
        NONE
    }
}

fn require_root_keys<T: ReadTxn>(txn: &T, expected: &[&str]) -> Result<(), String> {
    let actual = txn
        .root_refs()
        .map(|(key, _)| key.to_string())
        .collect::<BTreeSet<_>>();
    let expected = expected
        .iter()
        .map(|key| (*key).to_string())
        .collect::<BTreeSet<_>>();
    if actual == expected {
        Ok(())
    } else {
        Err("collaborative document roots do not match the schema".to_string())
    }
}

fn require_map_keys<T: ReadTxn>(
    map: &MapRef,
    txn: &T,
    expected: &[&str],
    label: &str,
) -> Result<(), String> {
    require_map_keys_with_optional(map, txn, expected, &[], label)
}

/// `optional` keys may be present or absent. A key two replicas assign at once
/// is deleted outright when the winning assignment is undone, so any such key
/// has to read as absent rather than as a broken document.
fn require_map_keys_with_optional<T: ReadTxn>(
    map: &MapRef,
    txn: &T,
    expected: &[&str],
    optional: &[&str],
    label: &str,
) -> Result<(), String> {
    let actual = map.keys(txn).map(str::to_string).collect::<BTreeSet<_>>();
    let expected = expected
        .iter()
        .map(|key| (*key).to_string())
        .collect::<BTreeSet<_>>();
    let optional = optional
        .iter()
        .map(|key| (*key).to_string())
        .collect::<BTreeSet<_>>();
    if expected.is_subset(&actual)
        && actual
            .difference(&expected)
            .all(|key| optional.contains(key))
    {
        Ok(())
    } else {
        Err(format!("{label} keys do not match the schema"))
    }
}

fn cell_key(at: CellRef) -> String {
    format!("{}:{}", at.row, at.col)
}

fn parse_cell_key(key: &str) -> Result<CellRef, String> {
    let (row, col) = key
        .split_once(':')
        .ok_or_else(|| format!("invalid cell key {key}"))?;
    let row = row
        .parse::<u32>()
        .map_err(|_| format!("invalid cell row in {key}"))?;
    let col = col
        .parse::<u32>()
        .map_err(|_| format!("invalid cell column in {key}"))?;
    let at = CellRef::new(row, col);
    if key != cell_key(at) {
        return Err(format!("noncanonical cell key {key}"));
    }
    if row >= MAX_ROWS || col >= MAX_COLS {
        return Err(format!("cell key {key} is out of bounds"));
    }
    Ok(at)
}

fn content_to_any(cell: &Cell) -> Option<Any> {
    if let Some(formula) = &cell.formula {
        Some(any_array(vec![
            Any::BigInt(1),
            Any::from(formula.as_str()),
            value_to_any(&cell.value),
        ]))
    } else if !matches!(&cell.value, CellValue::Empty) {
        Some(any_array(vec![Any::BigInt(0), value_to_any(&cell.value)]))
    } else {
        None
    }
}

fn content_from_any(value: &Any) -> Result<Cell, String> {
    let values = any_values(value, "cell content")?;
    let kind = values
        .first()
        .ok_or_else(|| "cell content is empty".to_string())?;
    match any_i64(kind, "cell content kind")? {
        0 if values.len() == 2 => {
            let value = value_from_any(&values[1])?;
            if matches!(&value, CellValue::Empty) {
                return Err("empty literal cell content must be omitted".to_string());
            }
            Ok(Cell {
                value,
                ..Cell::default()
            })
        }
        1 if values.len() == 3 => {
            let Any::String(formula) = &values[1] else {
                return Err("formula cell content is missing formula text".to_string());
            };
            Ok(Cell {
                value: value_from_any(&values[2])?,
                formula: Some(formula.to_string()),
                ..Cell::default()
            })
        }
        0 | 1 => Err("cell content has the wrong payload length".to_string()),
        kind => Err(format!("unsupported cell content kind {kind}")),
    }
}

fn value_to_any(value: &CellValue) -> Any {
    match value {
        CellValue::Empty => any_array(vec![Any::BigInt(0)]),
        CellValue::Number { value } => any_array(vec![Any::BigInt(1), Any::Number(*value)]),
        CellValue::Text { value } => any_array(vec![Any::BigInt(2), Any::from(value.as_str())]),
        CellValue::Bool { value } => any_array(vec![Any::BigInt(3), Any::Bool(*value)]),
        CellValue::Error { value } => any_array(vec![Any::BigInt(4), Any::from(value.as_str())]),
    }
}

fn value_from_any(value: &Any) -> Result<CellValue, String> {
    let values = any_values(value, "cell value")?;
    let kind = values
        .first()
        .ok_or_else(|| "cell value is empty".to_string())?;
    match any_i64(kind, "cell value kind")? {
        0 if values.len() == 1 => Ok(CellValue::Empty),
        1 if values.len() == 2 => match &values[1] {
            Any::Number(value) if value.is_finite() => Ok(CellValue::Number { value: *value }),
            _ => Err("numeric cell has a non-number value".to_string()),
        },
        2 if values.len() == 2 => match &values[1] {
            Any::String(value) => Ok(CellValue::Text {
                value: value.to_string(),
            }),
            _ => Err("text cell has a non-string value".to_string()),
        },
        3 if values.len() == 2 => match &values[1] {
            Any::Bool(value) => Ok(CellValue::Bool { value: *value }),
            _ => Err("boolean cell has a non-boolean value".to_string()),
        },
        4 if values.len() == 2 => match &values[1] {
            Any::String(value) => Ok(CellValue::Error {
                value: error_from_str(value)?,
            }),
            _ => Err("error cell has a non-string value".to_string()),
        },
        0..=4 => Err("cell value has the wrong payload length".to_string()),
        kind => Err(format!("unsupported cell value kind {kind}")),
    }
}

fn any_array(values: Vec<Any>) -> Any {
    Any::Array(Arc::from(values))
}

fn any_values<'a>(value: &'a Any, label: &str) -> Result<&'a [Any], String> {
    match value {
        Any::Array(values) => Ok(values),
        _ => Err(format!("{label} is not an array")),
    }
}

fn any_i64(value: &Any, label: &str) -> Result<i64, String> {
    match value {
        Any::BigInt(value) => Ok(*value),
        _ => Err(format!("{label} is not an integer")),
    }
}

fn error_from_str(value: &str) -> Result<ErrorValue, String> {
    match value {
        "#DIV/0!" => Ok(ErrorValue::Div0),
        "#N/A" => Ok(ErrorValue::NA),
        "#NAME?" => Ok(ErrorValue::Name),
        "#NULL!" => Ok(ErrorValue::Null),
        "#NUM!" => Ok(ErrorValue::Num),
        "#REF!" => Ok(ErrorValue::Ref),
        "#VALUE!" => Ok(ErrorValue::Value),
        "#SPILL!" => Ok(ErrorValue::Spill),
        "#CALC!" => Ok(ErrorValue::Calc),
        _ => Err(format!("unsupported cell error {value}")),
    }
}

fn merges_to_any(merges: &[CellRange]) -> Any {
    Any::Array(Arc::from(
        merges
            .iter()
            .map(|range| {
                any_array(vec![
                    Any::from(range.start.row),
                    Any::from(range.start.col),
                    Any::Bool(range.start.abs_row),
                    Any::Bool(range.start.abs_col),
                    Any::from(range.end.row),
                    Any::from(range.end.col),
                    Any::Bool(range.end.abs_row),
                    Any::Bool(range.end.abs_col),
                ])
            })
            .collect::<Vec<_>>(),
    ))
}

fn freeze_pane_to_any(pane: Option<FreezePane>) -> Any {
    match pane {
        None => Any::Null,
        Some(pane) => any_array(vec![
            Any::from(pane.rows),
            Any::from(pane.cols),
            Any::from(pane.top_left.row),
            Any::from(pane.top_left.col),
            Any::Bool(pane.top_left.abs_row),
            Any::Bool(pane.top_left.abs_col),
        ]),
    }
}

fn freeze_pane_from_any(value: &Any) -> Result<Option<FreezePane>, String> {
    let Any::Array(values) = value else {
        return if matches!(value, Any::Null) {
            Ok(None)
        } else {
            Err("sheet freeze pane is not an array".to_string())
        };
    };
    if values.len() != 6 {
        return Err("sheet freeze pane must contain six values".to_string());
    }
    let pane = FreezePane::new(
        any_u32(&values[0], "frozen row count")?,
        any_u32(&values[1], "frozen column count")?,
        CellRef {
            row: any_u32(&values[2], "freeze pane top row")?,
            col: any_u32(&values[3], "freeze pane left column")?,
            abs_row: any_bool(&values[4], "freeze pane absolute row")?,
            abs_col: any_bool(&values[5], "freeze pane absolute column")?,
        },
    );
    if pane.rows > MAX_ROWS
        || pane.cols > MAX_COLS
        || pane.top_left.row >= MAX_ROWS
        || pane.top_left.col >= MAX_COLS
    {
        return Err("sheet freeze pane is out of bounds".to_string());
    }
    Ok(Some(pane))
}

fn merges_from_any(value: &Any) -> Result<Vec<CellRange>, String> {
    let Any::Array(merges) = value else {
        return Err("sheet merges are not an array".to_string());
    };
    merges
        .iter()
        .map(|merge| {
            let Any::Array(values) = merge else {
                return Err("merge entry is not an array".to_string());
            };
            if values.len() != 8 {
                return Err("merge entry must contain eight values".to_string());
            }
            Ok(CellRange {
                start: CellRef {
                    row: any_u32(&values[0], "merge start row")?,
                    col: any_u32(&values[1], "merge start column")?,
                    abs_row: any_bool(&values[2], "merge start absolute row")?,
                    abs_col: any_bool(&values[3], "merge start absolute column")?,
                },
                end: CellRef {
                    row: any_u32(&values[4], "merge end row")?,
                    col: any_u32(&values[5], "merge end column")?,
                    abs_row: any_bool(&values[6], "merge end absolute row")?,
                    abs_col: any_bool(&values[7], "merge end absolute column")?,
                },
            })
        })
        .collect()
}

fn any_u32(value: &Any, label: &str) -> Result<u32, String> {
    match value {
        Any::Number(value)
            if value.is_finite()
                && *value >= 0.0
                && *value <= u32::MAX as f64
                && value.fract() == 0.0 =>
        {
            Ok(*value as u32)
        }
        Any::BigInt(value) if *value >= 0 && *value <= i64::from(u32::MAX) => Ok(*value as u32),
        _ => Err(format!("{label} is not a u32")),
    }
}

fn any_bool(value: &Any, label: &str) -> Result<bool, String> {
    match value {
        Any::Bool(value) => Ok(*value),
        _ => Err(format!("{label} is not a boolean")),
    }
}

/// Dimensions only this build's parser records: a hidden row or column with no
/// authored size. A state written before it did carries no entry at all, so
/// materializing one has to put them back or the row silently unhides.
#[derive(Clone, Debug, Default)]
struct HiddenDimensions {
    col_widths: BTreeMap<u32, f64>,
    row_heights: BTreeMap<u32, f64>,
}

static EMPTY_HIDDEN_DIMENSIONS: HiddenDimensions = HiddenDimensions {
    col_widths: BTreeMap::new(),
    row_heights: BTreeMap::new(),
};

fn hidden_dimensions(
    model: &WorkbookModel,
    legacy_dimensions: &[xlsx_parse::LegacySheetDimensions],
) -> Vec<HiddenDimensions> {
    model
        .sheets
        .iter()
        .enumerate()
        .map(|(index, sheet)| {
            let Some(legacy) = legacy_dimensions.get(index) else {
                return HiddenDimensions::default();
            };
            let only_current = |current: &BTreeMap<u32, f64>, before: &BTreeMap<u32, f64>| {
                current
                    .iter()
                    .filter(|(at, _)| !before.contains_key(at))
                    .map(|(at, size)| (*at, *size))
                    .collect()
            };
            HiddenDimensions {
                col_widths: only_current(&sheet.col_widths, &legacy.col_widths),
                row_heights: only_current(&sheet.row_heights, &legacy.row_heights),
            }
        })
        .collect()
}

/// The same workbook as an earlier release would have modelled it, or `None`
/// when no sheet's dimensions changed meaning and the two agree already.
fn model_with_legacy_dimensions(
    model: &WorkbookModel,
    legacy_dimensions: &[xlsx_parse::LegacySheetDimensions],
) -> Option<WorkbookModel> {
    let differs = model
        .sheets
        .iter()
        .zip(legacy_dimensions)
        .any(|(sheet, legacy)| {
            sheet.col_widths != legacy.col_widths || sheet.row_heights != legacy.row_heights
        });
    if !differs {
        return None;
    }
    let mut legacy_model = model.clone();
    for (sheet, legacy) in legacy_model.sheets.iter_mut().zip(legacy_dimensions) {
        sheet.col_widths.clone_from(&legacy.col_widths);
        sheet.row_heights.clone_from(&legacy.row_heights);
    }
    Some(legacy_model)
}

fn fingerprint_model(model: &WorkbookModel) -> Result<(String, u64), String> {
    fingerprint_model_for_schema(model, SCHEMA_VERSION)
}

fn fingerprint_model_for_schema(
    model: &WorkbookModel,
    schema_version: i64,
) -> Result<(String, u64), String> {
    fingerprint_model_with_schema(model, schema_version, schema_version >= 4, true)
}

fn fingerprint_model_with_schema(
    model: &WorkbookModel,
    schema_version: i64,
    include_defined_names: bool,
    include_col_styles: bool,
) -> Result<(String, u64), String> {
    validate_schema_version(schema_version)?;
    let mut hasher = Sha256::new();
    let domain = match schema_version {
        3 => b"betteroffice-xlsx-yrs-v3".as_slice(),
        4 => b"betteroffice-xlsx-yrs-v4".as_slice(),
        5 => b"betteroffice-xlsx-yrs-v5".as_slice(),
        _ => b"betteroffice-xlsx-yrs-v6".as_slice(),
    };
    hasher.update(domain);
    let base = if include_defined_names {
        serde_json::to_vec(&(
            model.date_system,
            &model.defined_names,
            &model.shared_strings,
            &model.styles,
        ))
    } else {
        serde_json::to_vec(&(model.date_system, &model.shared_strings, &model.styles))
    }
    .map_err(|error| format!("cannot fingerprint workbook base: {error}"))?;
    hash_bytes(&mut hasher, &base);
    hash_u64(&mut hasher, model.sheets.len() as u64);
    // a workbook declaring no column style hashes exactly as releases before
    // they were read did, so only a workbook that declares one needs the
    // pre-change fingerprint carried alongside.
    let hash_col_styles = include_col_styles
        && schema_version >= CHARTS_SCHEMA_VERSION
        && model
            .sheets
            .iter()
            .any(|sheet| !sheet.col_styles.is_empty());
    for sheet in &model.sheets {
        hash_bytes(&mut hasher, sheet.name.as_bytes());
        hash_u64(&mut hasher, sheet.iter_cells().count() as u64);
        for (at, cell) in sheet.iter_cells() {
            hash_u32(&mut hasher, at.row);
            hash_u32(&mut hasher, at.col);
            hash_cell_value(&mut hasher, &cell.value);
            match &cell.formula {
                Some(formula) => {
                    hasher.update([1]);
                    hash_bytes(&mut hasher, formula.as_bytes());
                }
                None => hasher.update([0]),
            }
            match cell.style {
                Some(style) => {
                    hasher.update([1]);
                    hash_u32(&mut hasher, style);
                }
                None => hasher.update([0]),
            }
        }
        hash_u64(&mut hasher, sheet.merges.len() as u64);
        for range in &sheet.merges {
            hash_cell_ref(&mut hasher, range.start);
            hash_cell_ref(&mut hasher, range.end);
        }
        hash_u64(&mut hasher, sheet.col_widths.len() as u64);
        for (&column, &width) in &sheet.col_widths {
            hash_u32(&mut hasher, column);
            hash_u64(&mut hasher, width.to_bits());
        }
        hash_u64(&mut hasher, sheet.row_heights.len() as u64);
        for (&row, &height) in &sheet.row_heights {
            hash_u32(&mut hasher, row);
            hash_u64(&mut hasher, height.to_bits());
        }
        if schema_version >= FREEZE_PANE_SCHEMA_VERSION {
            match sheet.freeze_pane {
                Some(pane) => {
                    hasher.update([1]);
                    hash_u32(&mut hasher, pane.rows);
                    hash_u32(&mut hasher, pane.cols);
                    hash_cell_ref(&mut hasher, pane.top_left);
                }
                None => hasher.update([0]),
            }
        }
        if schema_version >= HYPERLINK_SCHEMA_VERSION {
            let hyperlinks = serde_json::to_vec(&sheet.hyperlinks)
                .map_err(|error| format!("cannot fingerprint sheet hyperlinks: {error}"))?;
            hash_bytes(&mut hasher, &hyperlinks);
        }
        if schema_version >= CHARTS_SCHEMA_VERSION {
            let charts = serde_json::to_vec(&sheet.charts)
                .map_err(|error| format!("cannot fingerprint sheet charts: {error}"))?;
            hash_bytes(&mut hasher, &charts);
        }
        if hash_col_styles {
            let col_styles = serde_json::to_vec(&sheet.col_styles)
                .map_err(|error| format!("cannot fingerprint sheet column styles: {error}"))?;
            hash_bytes(&mut hasher, &col_styles);
        }
    }
    let digest = hasher.finalize();
    let fingerprint = format!("{digest:x}");
    let mut client_bytes = [0_u8; 8];
    client_bytes.copy_from_slice(&digest[..8]);
    let mut bootstrap_client_id = u64::from_be_bytes(client_bytes) & MAX_SAFE_CLIENT_ID;
    if bootstrap_client_id == 0 {
        bootstrap_client_id = 1;
    }
    Ok((fingerprint, bootstrap_client_id))
}

fn hash_cell_value(hasher: &mut Sha256, value: &CellValue) {
    match value {
        CellValue::Empty => hasher.update([0]),
        CellValue::Number { value } => {
            hasher.update([1]);
            hash_u64(hasher, value.to_bits());
        }
        CellValue::Text { value } => {
            hasher.update([2]);
            hash_bytes(hasher, value.as_bytes());
        }
        CellValue::Bool { value } => hasher.update([3, u8::from(*value)]),
        CellValue::Error { value } => {
            hasher.update([4]);
            hash_bytes(hasher, value.as_str().as_bytes());
        }
    }
}

fn hash_cell_ref(hasher: &mut Sha256, cell: CellRef) {
    hash_u32(hasher, cell.row);
    hash_u32(hasher, cell.col);
    hasher.update([u8::from(cell.abs_row), u8::from(cell.abs_col)]);
}

fn hash_bytes(hasher: &mut Sha256, bytes: &[u8]) {
    hash_u64(hasher, bytes.len() as u64);
    hasher.update(bytes);
}

fn hash_u32(hasher: &mut Sha256, value: u32) {
    hasher.update(value.to_le_bytes());
}

fn hash_u64(hasher: &mut Sha256, value: u64) {
    hasher.update(value.to_le_bytes());
}

#[cfg(test)]
mod tests {
    use super::*;
    use xlsx_model::Xf;

    fn rich_model() -> WorkbookModel {
        let mut first = Sheet::new("Data");
        first.set_cell(
            CellRef::new(0, 0),
            Cell {
                value: CellValue::Number { value: 42.0 },
                formula: Some("40+2".into()),
                style: Some(0),
            },
        );
        first.set_cell(
            CellRef::new(1, 0),
            Cell {
                value: CellValue::Text {
                    value: "hello".into(),
                },
                ..Cell::default()
            },
        );
        first.col_widths.insert(1, 24.5);
        first.row_heights.insert(2, 30.0);
        first
            .merges
            .push(CellRange::new(CellRef::new(3, 0), CellRef::new(4, 2)));
        first.freeze_pane = Some(FreezePane::new(1, 1, CellRef::new(3, 2)));
        first.hyperlinks.push(Hyperlink {
            range: CellRange::new(CellRef::new(1, 0), CellRef::new(1, 0)),
            external_target: Some("https://example.com".into()),
            location: None,
            tooltip: Some("Open".into()),
            display: Some("Example".into()),
        });
        let mut model = WorkbookModel {
            date_system: DateSystem::V1904,
            shared_strings: vec!["hello".into()],
            ..WorkbookModel::default()
        };
        model.defined_names.push(DefinedName {
            name: "Answer".into(),
            formula: "Data!A1".into(),
            local_sheet: None,
            hidden: false,
        });
        model.styles.cell_xfs.push(Xf::default());
        model.sheets.push(first);
        model.sheets.push(Sheet::new("Second"));
        model
    }

    fn legacy_update(model: &WorkbookModel, version: i64, include_defined_names: bool) -> Vec<u8> {
        let base = WorkbookBase::from_model(model).unwrap();
        let (_, client_id) =
            fingerprint_model_with_schema(model, version, include_defined_names, true).unwrap();
        let doc = Doc::with_client_id(client_id);
        let keys = (0..model.sheets.len())
            .map(|index| format!("sheet:{index}"))
            .collect::<Vec<_>>();
        seed(&doc, &base, model, &keys).unwrap();
        {
            let (fingerprint, _) =
                fingerprint_model_with_schema(model, version, include_defined_names, true).unwrap();
            let mut txn = doc.transact_mut_with("test:legacy-schema");
            let meta = txn.get_map(META).unwrap();
            meta.try_update(&mut txn, BASE_FINGERPRINT, fingerprint);
            meta.try_update(&mut txn, "schemaVersion", version);
            let sheets = txn.get_map(SHEETS).unwrap();
            for key in keys {
                let sheet = sheets
                    .get(&txn, &key)
                    .and_then(|value| value.cast::<MapRef>().ok())
                    .unwrap();
                if version < CHARTS_SCHEMA_VERSION {
                    sheet.remove(&mut txn, CHARTS);
                }
                if version < HYPERLINK_SCHEMA_VERSION {
                    sheet.remove(&mut txn, HYPERLINKS);
                }
                if version < FREEZE_PANE_SCHEMA_VERSION {
                    sheet.remove(&mut txn, FREEZE_PANE);
                }
            }
        }
        doc.transact()
            .encode_state_as_update_v1(&StateVector::default())
    }

    fn authority_from_update(
        model: &WorkbookModel,
        update: &[u8],
        client_id: u64,
    ) -> WorkbookAuthority {
        let doc = Doc::with_client_id(client_id);
        hydrate_doc(&doc, update).unwrap();
        WorkbookAuthority {
            doc,
            base: Arc::new(WorkbookBase::from_model(model).unwrap()),
            history: SheetOrderHistory::default(),
            next_sheet_id: 0,
            undo_stack: Vec::new(),
            redo_stack: Vec::new(),
        }
    }

    #[test]
    fn a_column_styled_workbook_still_accepts_its_pre_change_fingerprint() {
        let mut model = rich_model();
        model.sheets[0].col_styles = vec![ColStyle {
            first: 0,
            last: 3,
            xf: 1,
        }];
        let base = WorkbookBase::from_model(&model).unwrap();
        for version in MIN_SUPPORTED_SCHEMA_VERSION..=SCHEMA_VERSION {
            let (pre_change, _) =
                fingerprint_model_with_schema(&model, version, version >= 4, false).unwrap();
            assert!(
                base.fingerprints[&version].contains(&pre_change),
                "schema v{version} must still recognise the fingerprint released before \
                 column styles were read"
            );
        }
        let (current, _) = fingerprint_model_for_schema(&model, SCHEMA_VERSION).unwrap();
        let (without, _) =
            fingerprint_model_with_schema(&model, SCHEMA_VERSION, true, false).unwrap();
        assert_ne!(
            current, without,
            "two workbooks differing only in their column styles must not share a fingerprint"
        );

        let mut plain = rich_model();
        plain.sheets[0].col_styles.clear();
        let (plain_current, _) = fingerprint_model_for_schema(&plain, SCHEMA_VERSION).unwrap();
        let (plain_without, _) =
            fingerprint_model_with_schema(&plain, SCHEMA_VERSION, true, false).unwrap();
        assert_eq!(
            plain_current, plain_without,
            "a workbook declaring no column style keeps the fingerprint it always had"
        );
    }

    #[test]
    fn deterministic_bootstrap_round_trips_formula_fallbacks() {
        let model = rich_model();
        let left = WorkbookAuthority::from_model_with_client_id(&model, 11).unwrap();
        let right = WorkbookAuthority::from_model_with_client_id(&model, 12).unwrap();
        assert_eq!(left.materialize().unwrap(), model);
        assert_eq!(right.materialize().unwrap(), model);
        assert_eq!(
            left.encode_state_vector_v1(),
            right.encode_state_vector_v1()
        );
        assert_eq!(
            left.encode_state_as_update_v1(),
            right.encode_state_as_update_v1()
        );
    }

    #[test]
    fn canonical_cell_formats_preserve_high_precision_theme_tints() {
        for tint in [
            "-0.14996795556505021",
            "-0.24994659260841701",
            "-4.9989318521683403E-2",
            "0.39994506668294322",
        ] {
            let format = CellFormat {
                fill: xlsx_model::Fill::Solid(xlsx_model::Color::Theme {
                    idx: 4,
                    tint: tint.parse().unwrap(),
                }),
                ..CellFormat::default()
            };
            let (key, payload) = cell_format_entry(&format).unwrap();
            let decoded: CellFormat = serde_json::from_str(&payload).unwrap();
            assert_eq!(decoded, format, "{tint}");
            assert_eq!(cell_format_entry(&decoded).unwrap(), (key, payload));

            let mut model = rich_model();
            let style = model.styles.intern_cell_format(&format).unwrap();
            model.sheets[0].set_cell(
                CellRef::new(1, 1),
                Cell {
                    value: CellValue::Number { value: 1.0 },
                    style,
                    ..Cell::default()
                },
            );
            let authority = WorkbookAuthority::from_model_with_client_id(&model, 11).unwrap();
            assert_eq!(authority.materialize().unwrap(), model);
            let peer = authority_from_update(&model, &authority.encode_state_as_update_v1(), 12);
            assert_eq!(peer.materialize().unwrap(), model);
        }
    }

    #[test]
    fn large_local_workbooks_stage_edits_and_restore_checkpoints() {
        let mut sheet = Sheet::new("Data");
        for row in 0..=(MAX_UPDATE_VALUES / 5) as u32 {
            sheet.set_cell(
                CellRef::new(row, 0),
                Cell {
                    value: CellValue::Number { value: row as f64 },
                    ..Cell::default()
                },
            );
        }
        let model = WorkbookModel {
            sheets: vec![sheet],
            ..WorkbookModel::default()
        };
        let mut authority = WorkbookAuthority::from_model_with_client_id(&model, 11).unwrap();
        let peer = WorkbookAuthority::from_model_with_client_id(&model, 12).unwrap();
        let snapshot = authority.encode_state_as_update_v1();
        assert!(decode_update_v1(&snapshot).is_err());
        assert!(matches!(
            peer.snapshot_replacement(&snapshot),
            SnapshotAdoption::NotApplicable
        ));
        assert!(authority.materialize().unwrap() == model);
        assert!(matches!(
            peer.stage_updates_v1(&[&snapshot], None),
            Err(AuthorityError::InvalidUpdate(_))
        ));
        let checkpoint = authority.checkpoint();

        let at = CellRef::new(0, 0);
        let staged = authority
            .stage_local_ops_v1(
                &[Op::SetCell {
                    sheet: SheetId(0),
                    at,
                    cell: xlsx_ops::CellState {
                        value: CellValue::Number { value: -1.0 },
                        ..xlsx_ops::CellState::default()
                    },
                }],
                SyncOrigin::User,
                &Stylesheet::default(),
            )
            .unwrap();
        authority
            .apply_local_update_v1(&staged.update, SyncOrigin::User)
            .unwrap();
        let remote = peer.stage_updates_v1(&[&staged.update], None).unwrap();
        peer.apply_staged_update_v1(&remote.commit_update).unwrap();
        assert!(peer.materialize().unwrap() == authority.materialize().unwrap());

        assert!(authority.undo().unwrap().unwrap().model == model);
        assert!(authority.redo().unwrap().unwrap().model == peer.materialize().unwrap());
        authority.restore(checkpoint).unwrap();
        assert!(!authority.can_undo());
        assert!(!authority.can_redo());
        assert!(authority.materialize().unwrap() == model);
    }

    #[test]
    fn indexed_palettes_survive_legacy_snapshots_and_distinguish_current_bases() {
        let mut model = rich_model();
        let legacy = WorkbookAuthority::from_model_with_client_id(&model, 11).unwrap();
        model.styles.indexed_colors = vec!["#123456".into(); 64];
        let current = WorkbookAuthority::from_model_with_client_id(&model, 12).unwrap();
        let SnapshotAdoption::Replacement(mut restored) =
            current.snapshot_replacement(&legacy.encode_state_as_update_v1())
        else {
            panic!("legacy snapshot should retain the source palette");
        };
        assert_eq!(restored.materialize().unwrap(), model);
        let mut other_model = model.clone();
        other_model.styles.indexed_colors[2] = "#abcdef".into();
        let other = WorkbookAuthority::from_model_with_client_id(&other_model, 13).unwrap();
        assert!(matches!(
            current.snapshot_replacement(&other.encode_state_as_update_v1()),
            SnapshotAdoption::Incompatible(_)
        ));
        restored
            .apply_ops(
                &[Op::SetCell {
                    sheet: SheetId(0),
                    at: CellRef::new(0, 0),
                    cell: xlsx_ops::CellState {
                        value: CellValue::Number { value: 99.0 },
                        ..Default::default()
                    },
                }],
                SyncOrigin::User,
                &Stylesheet::default(),
            )
            .unwrap();
        let restored_snapshot = restored.encode_state_as_update_v1();
        assert!(matches!(
            other.snapshot_replacement(&restored_snapshot),
            SnapshotAdoption::Incompatible(_)
        ));
        let SnapshotAdoption::Replacement(same_palette) =
            current.snapshot_replacement(&restored_snapshot)
        else {
            panic!("restored snapshots should match the source palette");
        };
        assert_eq!(
            same_palette.materialize().unwrap(),
            restored.materialize().unwrap()
        );
        assert!(!same_palette.upgrade_schema().unwrap());
    }

    #[test]
    fn known_schema_versions_materialize_and_upgrade_to_current() {
        let model = rich_model();
        for (index, (version, include_defined_names)) in
            [(3, false), (3, true), (4, true), (5, true)]
                .into_iter()
                .enumerate()
        {
            let update = legacy_update(&model, version, include_defined_names);
            let authority = authority_from_update(&model, &update, 101 + index as u64);
            assert_eq!(authority.strict_materialize().unwrap().0, model);

            let staged = authority
                .stage_updates_v1(&[Update::EMPTY_V1], None)
                .unwrap();
            assert!(staged.effective);
            authority
                .apply_staged_update_v1(&staged.commit_update)
                .unwrap();
            assert_eq!(authority.schema_version().unwrap(), SCHEMA_VERSION);
            assert_eq!(authority.strict_materialize().unwrap().0, model);
        }
    }

    #[test]
    fn legacy_snapshot_merges_into_current_bootstrap() {
        let model = rich_model();
        for (version, include_defined_names) in [(3, false), (3, true), (4, true), (5, true)] {
            let update = legacy_update(&model, version, include_defined_names);
            let authority = WorkbookAuthority::from_model_with_client_id(&model, 108).unwrap();
            let staged = authority.stage_updates_v1(&[&update], None).unwrap();
            assert_eq!(staged.model, model);
            authority
                .apply_staged_update_v1(&staged.commit_update)
                .unwrap();
            assert_eq!(authority.schema_version().unwrap(), SCHEMA_VERSION);
            assert_eq!(authority.strict_materialize().unwrap().0, model);
        }
    }

    /// The legacy fallback is keyed on `sheet:N`, not on where a sheet sits
    /// now, so a reordered legacy state still reads its own charts.
    #[test]
    fn a_reordered_legacy_state_keeps_each_sheet_its_own_charts() {
        let mut model = WorkbookModel::default();
        model.sheets.push(charted("First", "First!$A$1"));
        model.sheets.push(charted("Second", "Second!$A$1"));
        let base = WorkbookBase::from_model(&model).unwrap();
        let doc = Doc::with_client_id(base.bootstrap_client_id);
        seed(
            &doc,
            &base,
            &model,
            &["sheet:0".to_owned(), "sheet:1".to_owned()],
        )
        .unwrap();
        {
            let mut txn = doc.transact_mut_with("test:reordered-legacy");
            let order = txn.get_array(SHEET_ORDER).unwrap();
            order.remove(&mut txn, 0);
            order.insert(&mut txn, 1, "sheet:0");
            let sheets = txn.get_map(SHEETS).unwrap();
            for key in ["sheet:0", "sheet:1"] {
                let sheet = sheets
                    .get(&txn, key)
                    .and_then(|value| value.cast::<MapRef>().ok())
                    .unwrap();
                sheet.remove(&mut txn, CHARTS);
            }
            let (fingerprint, _) = fingerprint_model_with_schema(&model, 5, true, true).unwrap();
            let meta = txn.get_map(META).unwrap();
            meta.try_update(&mut txn, BASE_FINGERPRINT, fingerprint);
            meta.try_update(&mut txn, "schemaVersion", 5);
        }
        let update = doc
            .transact()
            .encode_state_as_update_v1(&StateVector::default());

        let mut uncharted = model.clone();
        for sheet in &mut uncharted.sheets {
            sheet.charts.clear();
        }
        let mut authority = authority_from_update(&uncharted, &update, 121);
        let fingerprints = authority.base.fingerprints.clone();
        authority.base = Arc::new(WorkbookBase::from_model(&model).unwrap());
        Arc::make_mut(&mut authority.base).fingerprints = fingerprints;

        let materialized = authority.materialize().unwrap();
        assert_eq!(materialized.sheets[0].name, "Second");
        assert_eq!(
            materialized.sheets[0].charts[0].refs[0].formula, "Second!$A$1",
            "the fallback must follow the stable key, not the position"
        );
        assert_eq!(materialized.sheets[1].name, "First");
        assert_eq!(
            materialized.sheets[1].charts[0].refs[0].formula,
            "First!$A$1"
        );
    }

    /// A v3-v5 state carries no chart state at all, so a charted workbook
    /// pairs with it on the rest and keeps the charts it parsed. Refusing
    /// instead would strand every snapshot written before charts were shared.
    #[test]
    fn a_charted_workbook_pairs_with_a_legacy_fingerprint() {
        let mut model = WorkbookModel::default();
        model.sheets.push(charted("Report", "Report!$A$1"));
        for version in MIN_SUPPORTED_SCHEMA_VERSION..SCHEMA_VERSION {
            let update = legacy_update(&model, version, true);
            let authority = authority_from_update(&model, &update, 130 + version as u64);
            let materialized = authority.materialize().unwrap();
            assert_eq!(materialized, model, "version {version}");
            assert!(authority.upgrade_schema().unwrap());
            assert_eq!(authority.strict_materialize().unwrap().0, model);
        }
    }

    /// Charts must stay pinned to the schema that introduced them. Gating them
    /// on the current version instead reads the newest schema as pre-chart the
    /// moment the current version moves past it, which is how a released
    /// snapshot came to be unreadable in the first place.
    ///
    /// The key is optional rather than required, because repinning is an
    /// ordinary edit and undoing the assignment that won a race deletes it. An
    /// absent key falls back to the source package, so the schema gate is what
    /// decides whether the document's own chart state is read at all.
    #[test]
    fn chart_state_is_gated_on_the_schema_that_introduced_it() {
        const { assert!(CHARTS_SCHEMA_VERSION <= SCHEMA_VERSION) };
        const { assert!(HYPERLINK_SCHEMA_VERSION < CHARTS_SCHEMA_VERSION) };

        let mut model = WorkbookModel::default();
        model.sheets.push(charted("Report", "Report!$A$1"));
        let authority = WorkbookAuthority::from_model_with_client_id(&model, 140).unwrap();

        // present: the document's own chart state wins over the fallback.
        let mut moved = model.sheets[0].charts.clone();
        moved[0].refs[0].formula = "Report!$B$9".to_owned();
        {
            let mut txn = authority.doc.transact_mut_with("test:charts-gate");
            let sheets = txn.get_map(SHEETS).unwrap();
            let sheet = sheets
                .get(&txn, "sheet:0")
                .and_then(|value| value.cast::<MapRef>().ok())
                .unwrap();
            sheet.try_update(
                &mut txn,
                CHARTS,
                serde_json::to_string(&moved).unwrap().as_str(),
            );
        }
        assert_eq!(
            authority.materialize().unwrap().sheets[0].charts[0].refs[0].formula,
            "Report!$B$9",
            "a document at the chart schema must read its own chart state"
        );

        // absent: the source package answers, and the chart is still there.
        {
            let mut txn = authority.doc.transact_mut_with("test:charts-gate");
            let sheets = txn.get_map(SHEETS).unwrap();
            let sheet = sheets
                .get(&txn, "sheet:0")
                .and_then(|value| value.cast::<MapRef>().ok())
                .unwrap();
            sheet.remove(&mut txn, CHARTS);
        }
        let fallen_back = authority
            .materialize()
            .expect("an absent chart key must read as absent, not as a broken workbook");
        assert_eq!(fallen_back.sheets[0].charts, model.sheets[0].charts);

        let gated = |version: i64| {
            sheet_schema_keys(version).contains(&CHARTS)
                || sheet_schema_optional_keys(version).contains(&CHARTS)
        };
        assert!(gated(CHARTS_SCHEMA_VERSION));
        assert!(!gated(CHARTS_SCHEMA_VERSION - 1));
    }

    #[test]
    fn unknown_schema_version_reports_supported_range() {
        let model = rich_model();
        let authority = WorkbookAuthority::from_model_with_client_id(&model, 110).unwrap();
        {
            let mut txn = authority.doc.transact_mut_with("test:unknown-schema");
            let meta = txn.get_map(META).unwrap();
            meta.try_update(&mut txn, "schemaVersion", SCHEMA_VERSION + 1);
        }
        let AuthorityError::InvalidState(error) = authority.materialize().unwrap_err() else {
            panic!("expected invalid state");
        };
        assert_eq!(
            error,
            "unsupported schema version 7; supported versions are 3 through 6"
        );
    }

    fn charted(name: &str, formula: &str) -> Sheet {
        let mut sheet = Sheet::new(name);
        sheet.charts.push(xlsx_model::SheetChart {
            part: "xl/charts/chart1.xml".to_owned(),
            drawing: "xl/drawings/drawing1.xml".to_owned(),
            anchor_index: 0,
            anchor: xlsx_model::ChartAnchor::Absolute {
                pos: xlsx_model::AnchorPos::default(),
                extent: xlsx_model::AnchorExtent::default(),
            },
            refs: vec![xlsx_model::ChartRef {
                kind: xlsx_model::ChartRefKind::Values,
                formula: formula.to_owned(),
            }],
        });
        sheet
    }

    /// Removing a sheet strands the chart references that named it, so the
    /// remaining sheets' chart state must reach the shared document. A partial
    /// sync would leave a peer reading the pre-removal ranges.
    #[test]
    fn removing_a_sheet_carries_the_stranded_chart_state_into_the_document() {
        let mut model = WorkbookModel::default();
        model.sheets.push(Sheet::new("Data"));
        model.sheets.push(charted("Report", "Data!$A$1:$A$2"));
        let mut authority = WorkbookAuthority::from_model(&model).unwrap();

        authority
            .apply_ops(
                &[Op::RemoveSheet { index: 0 }],
                SyncOrigin::User,
                &Stylesheet::default(),
            )
            .unwrap();

        let shared = authority.materialize().unwrap();
        assert_eq!(shared.sheets.len(), 1);
        assert_eq!(shared.sheets[0].charts[0].refs[0].formula, "#REF!");
    }

    /// `SetCharts` replaces a whole sheet's chart state, so it can only travel
    /// as a full semantic sync.
    #[test]
    fn set_charts_travels_as_a_full_semantic_sync() {
        let mut model = WorkbookModel::default();
        model.sheets.push(charted("Report", "Data!$A$1:$A$2"));
        let mut authority = WorkbookAuthority::from_model(&model).unwrap();

        let mut charts = model.sheets[0].charts.clone();
        charts[0].refs[0].formula = "Data!$A$1:$A$9".to_owned();
        authority
            .apply_ops(
                &[Op::SetCharts {
                    sheet: SheetId(0),
                    charts,
                }],
                SyncOrigin::User,
                &Stylesheet::default(),
            )
            .expect("SetCharts is a full sync, not a rejected partial one");

        let shared = authority.materialize().unwrap();
        assert_eq!(shared.sheets[0].charts[0].refs[0].formula, "Data!$A$1:$A$9");
    }

    #[test]
    fn oversized_peer_chart_state_is_refused_and_leaves_the_authority_intact() {
        const ELEMENT: &str = r#"{"part":"a","drawing":"b","anchorIndex":0,"anchor":{"kind":"absolute","pos":{"x":0,"y":0},"extent":{"cx":0,"cy":0}},"refs":[]}"#;
        let model = rich_model();
        for (client_id, count, expected) in [
            (111, 10_000, "sheet has too many charts"),
            (112, 80_000, "sheet chart state exceeds its size limit"),
        ] {
            let authority =
                WorkbookAuthority::from_model_with_client_id(&model, client_id).unwrap();
            let peer = Doc::with_client_id(client_id + 1);
            hydrate_doc(&peer, &authority.encode_state_as_update_v1()).unwrap();
            let before = peer.transact().state_vector();
            let mut payload = String::with_capacity(count * (ELEMENT.len() + 1) + 2);
            payload.push('[');
            for index in 0..count {
                if index > 0 {
                    payload.push(',');
                }
                payload.push_str(ELEMENT);
            }
            payload.push(']');
            {
                let mut txn = peer.transact_mut_with("test:hostile-charts");
                let sheets = txn.get_map(SHEETS).unwrap();
                let sheet = sheets
                    .get(&txn, "sheet:0")
                    .and_then(|value| value.cast::<MapRef>().ok())
                    .unwrap();
                sheet.try_update(&mut txn, CHARTS, payload.as_str());
            }
            let update = peer.transact().encode_diff_v1(&before);
            let Err(AuthorityError::InvalidState(error)) =
                authority.stage_updates_v1(&[&update], None)
            else {
                panic!("expected invalid state");
            };
            assert!(error.contains(expected), "{error}");
            assert_eq!(authority.strict_materialize().unwrap().0, model);
        }
    }

    #[test]
    fn formula_content_is_one_atomic_payload() {
        let formula = Cell {
            value: CellValue::Error {
                value: ErrorValue::Ref,
            },
            formula: Some("Missing!A1".into()),
            style: None,
        };
        assert_eq!(
            content_from_any(&content_to_any(&formula).unwrap()).unwrap(),
            formula
        );
    }

    #[test]
    fn strict_decoders_reject_trailing_and_impossible_vectors() {
        assert!(decode_state_vector_v1(&[1]).is_err());
        assert!(decode_state_vector_v1(&[0, 0]).is_err());
        assert!(decode_update_v1(&[0, 0, 0]).is_err());
    }

    #[test]
    fn update_decoder_rejects_overflowing_block_and_delete_clocks() {
        let overflowing_skip = [
            1,
            1,
            1,
            0xff,
            0xff,
            0xff,
            0xff,
            0x0f,
            BLOCK_SKIP_REF_NUMBER,
            1,
            0,
        ];
        let overflowing_gc = [
            1,
            1,
            1,
            0x80,
            0x80,
            0x80,
            0x80,
            0x08,
            BLOCK_GC_REF_NUMBER,
            1,
            0,
        ];
        let overflowing_delete = [0, 1, 1, 1, 0xff, 0xff, 0xff, 0xff, 0x0f, 1];
        assert!(decode_update_v1(&overflowing_skip).is_err());
        assert!(decode_update_v1(&overflowing_gc).is_err());
        assert!(decode_update_v1(&overflowing_delete).is_err());
    }

    #[test]
    fn checked_decoders_bound_counts_and_reject_noncanonical_ids() {
        let too_many_blocks = [1, 0xff, 0xff, 0xff, 0xff, 0x0f];
        let too_many_delete_ranges = [0, 1, 1, 0xff, 0xff, 0xff, 0xff, 0x0f];
        assert!(decode_update_v1(&too_many_blocks).is_err());
        assert!(decode_update_v1(&too_many_delete_ranges).is_err());
        assert!(decode_state_vector_v1(&[1, 0x81, 0x80, 0, 0]).is_err());
    }

    #[test]
    fn checked_decoder_rejects_overlong_signed_varints() {
        for bytes in [
            vec![125, 0x40],
            vec![125, 0x80, 0],
            vec![125, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0],
        ] {
            assert!(CheckedDecoderV1::new(&bytes).decode_any(0).is_err());
        }
        assert_eq!(
            CheckedDecoderV1::new(&[125, 0xc1, 1])
                .decode_any(0)
                .unwrap(),
            Any::Number(-65.0)
        );
    }

    #[test]
    fn state_vector_entry_count_is_bounded_before_decode() {
        let error = decode_state_vector_v1(&[0x81, 0x80, 0x04]).unwrap_err();
        assert!(error.contains("65536-entry limit"), "{error}");
    }

    #[test]
    fn bootstrap_client_id_conflicts_are_rejected_locally() {
        let model = rich_model();
        let base = WorkbookBase::from_model(&model).unwrap();
        assert!(matches!(
            WorkbookAuthority::from_model_with_client_id(&model, base.bootstrap_client_id),
            Err(AuthorityError::ClientIdConflict(_))
        ));
    }

    #[test]
    fn shared_map_replacement_changes_the_frozen_structure() {
        let model = rich_model();
        let source = WorkbookAuthority::from_model_with_client_id(&model, 21).unwrap();
        let target = WorkbookAuthority::from_model_with_client_id(&model, 22).unwrap();
        let target_structure = target.structure().unwrap();
        let target_vector = target.encode_state_vector_v1();

        {
            let mut txn = source.doc.transact_mut_with("test:replace-map");
            let sheets = txn.get_map(SHEETS).unwrap();
            let sheet = sheets
                .get(&txn, "sheet:1")
                .and_then(|value| value.cast::<MapRef>().ok())
                .unwrap();
            sheet.insert(&mut txn, CONTENTS, MapPrelim::default());
        }

        let update = source.encode_diff_v1(&target_vector).unwrap();
        let staged = target.stage_updates_v1(&[&update], None).unwrap();
        assert_eq!(staged.model, target.materialize().unwrap());
        assert_ne!(staged.structure, target_structure);
    }

    #[test]
    fn retained_sheet_maps_stay_valid_and_keep_identity_through_undo() {
        let model = rich_model();
        let mut authority = WorkbookAuthority::from_model_with_client_id(&model, 31).unwrap();
        authority
            .apply_ops(
                &[Op::RemoveSheet { index: 1 }],
                SyncOrigin::User,
                &Stylesheet::default(),
            )
            .unwrap();
        let (_, removed) = authority.strict_materialize().unwrap();
        assert_eq!(removed.sheet_keys, ["sheet:0"]);
        assert_eq!(removed.shared_types.len(), 2);
        let retained = removed.shared_types["sheet:1"].clone();

        authority
            .apply_ops(
                &[Op::AddSheet {
                    index: 1,
                    name: "Second".into(),
                }],
                SyncOrigin::Undo,
                &Stylesheet::default(),
            )
            .unwrap();
        let (restored, structure) = authority.strict_materialize().unwrap();
        assert_eq!(restored, model);
        assert_eq!(structure.shared_types["sheet:1"], retained);
    }

    fn sliding_chart(name: &str) -> Sheet {
        let mut sheet = Sheet::new(name);
        sheet.charts.push(SheetChart {
            part: "xl/charts/chart1.xml".to_owned(),
            drawing: "xl/drawings/drawing1.xml".to_owned(),
            anchor_index: 0,
            anchor: ChartAnchor::TwoCell {
                from: xlsx_model::AnchorCell::default(),
                to: xlsx_model::AnchorCell {
                    col: 4,
                    col_off: 0,
                    row: 8,
                    row_off: 0,
                },
                edit_as: AnchorEditAs::TwoCell,
            },
            refs: vec![ChartRef {
                kind: xlsx_model::ChartRefKind::Values,
                formula: "Data!$A$1:$A$2".to_owned(),
            }],
        });
        sheet
    }

    fn sliding_model() -> WorkbookModel {
        let mut model = WorkbookModel::default();
        model.sheets.push(sliding_chart("Report"));
        model
    }

    /// An update that assigns a sheet's chart state without touching the
    /// structure generation, as a peer forked from `authority`'s state.
    fn peer_chart_update(authority: &WorkbookAuthority, client_id: u64, charts: &str) -> Vec<u8> {
        let peer = Doc::with_client_id(client_id);
        hydrate_doc(&peer, &authority.encode_state_as_update_v1()).unwrap();
        let before = peer.transact().state_vector();
        {
            let mut txn = peer.transact_mut_with("test:peer-charts");
            let sheets = txn.get_map(SHEETS).unwrap();
            let sheet = sheets
                .get(&txn, "sheet:0")
                .and_then(|value| value.cast::<MapRef>().ok())
                .unwrap();
            sheet.try_update(&mut txn, CHARTS, charts);
        }
        peer.transact().encode_diff_v1(&before)
    }

    fn slid_anchor(cols: i64) -> ChartAnchor {
        ChartAnchor::TwoCell {
            from: xlsx_model::AnchorCell {
                col: cols as u32,
                ..xlsx_model::AnchorCell::default()
            },
            to: xlsx_model::AnchorCell {
                col: 4 + cols as u32,
                col_off: 0,
                row: 8,
                row_off: 0,
            },
            edit_as: AnchorEditAs::TwoCell,
        }
    }

    /// The freeze pins what a chart *is*, so a peer cannot pass a remap off as
    /// a move. The structure generation is untouched here, which is what makes
    /// this the identity check rather than the generation counter.
    #[test]
    fn a_peer_cannot_disguise_a_chart_remap_as_a_move() {
        let model = sliding_model();
        let authority = WorkbookAuthority::from_model_with_client_id(&model, 41).unwrap();
        let frozen = authority.structure().unwrap();
        let generation = frozen.generation;

        for (label, charts) in [
            (
                "refs",
                r#"[{"part":"xl/charts/chart1.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":0,"anchor":{"kind":"twoCell","from":{"col":0,"colOff":0,"row":0,"rowOff":0},"to":{"col":4,"colOff":0,"row":8,"rowOff":0},"edit_as":"twoCell"},"refs":[{"kind":"values","formula":"Hijacked!$A$1"}]}]"#,
            ),
            (
                "part",
                r#"[{"part":"xl/charts/other.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":0,"anchor":{"kind":"twoCell","from":{"col":0,"colOff":0,"row":0,"rowOff":0},"to":{"col":4,"colOff":0,"row":8,"rowOff":0},"edit_as":"twoCell"},"refs":[{"kind":"values","formula":"Data!$A$1:$A$2"}]}]"#,
            ),
            (
                "anchorIndex",
                r#"[{"part":"xl/charts/chart1.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":3,"anchor":{"kind":"twoCell","from":{"col":0,"colOff":0,"row":0,"rowOff":0},"to":{"col":4,"colOff":0,"row":8,"rowOff":0},"edit_as":"twoCell"},"refs":[{"kind":"values","formula":"Data!$A$1:$A$2"}]}]"#,
            ),
        ] {
            let update = peer_chart_update(&authority, 42, charts);
            let staged = authority.stage_updates_v1(&[&update], None).unwrap();
            assert_eq!(
                staged.structure.generation, generation,
                "{label} must not move the generation, or it proves nothing"
            );
            assert_ne!(
                staged.structure, frozen,
                "a rewritten {label} must change the frozen structure"
            );
        }
    }

    /// A move may slide a grid-anchored chart and nothing else. The drawing
    /// writer refuses a changed kind, `editAs` mode or one-cell extent, so the
    /// freeze has to refuse them too rather than accept a workbook that can no
    /// longer be saved.
    #[test]
    fn a_peer_cannot_reshape_an_anchor_a_save_can_only_slide() {
        let model = sliding_model();
        let authority = WorkbookAuthority::from_model_with_client_id(&model, 43).unwrap();
        let frozen = authority.structure().unwrap();

        for (label, charts) in [
            (
                "editAs",
                r#"[{"part":"xl/charts/chart1.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":0,"anchor":{"kind":"twoCell","from":{"col":0,"colOff":0,"row":0,"rowOff":0},"to":{"col":4,"colOff":0,"row":8,"rowOff":0},"edit_as":"oneCell"},"refs":[{"kind":"values","formula":"Data!$A$1:$A$2"}]}]"#,
            ),
            (
                "kind",
                r#"[{"part":"xl/charts/chart1.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":0,"anchor":{"kind":"oneCell","from":{"col":0,"colOff":0,"row":0,"rowOff":0},"extent":{"cx":100000,"cy":100000}},"refs":[{"kind":"values","formula":"Data!$A$1:$A$2"}]}]"#,
            ),
        ] {
            let update = peer_chart_update(&authority, 44, charts);
            let staged = authority.stage_updates_v1(&[&update], None).unwrap();
            assert_ne!(
                staged.structure, frozen,
                "a rewritten anchor {label} must change the frozen structure"
            );
        }

        // sliding the same anchor across the grid is the one accepted change.
        let slid = r#"[{"part":"xl/charts/chart1.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":0,"anchor":{"kind":"twoCell","from":{"col":2,"colOff":0,"row":0,"rowOff":0},"to":{"col":6,"colOff":0,"row":8,"rowOff":0},"edit_as":"twoCell"},"refs":[{"kind":"values","formula":"Data!$A$1:$A$2"}]}]"#;
        let update = peer_chart_update(&authority, 44, slid);
        let staged = authority.stage_updates_v1(&[&update], None).unwrap();
        assert_eq!(staged.structure, frozen);
    }

    /// A checkpoint has to put back everything a history step moves: the
    /// document, both stacks, and the identity the stacks name. Restoring into
    /// a fresh document would look right and leave the history unusable — the
    /// undo manager drops entries belonging to a document it does not know,
    /// silently, a whole stack at a time — so the proof is that undo still
    /// works afterwards.
    #[test]
    fn a_restored_checkpoint_brings_back_a_working_history() {
        let model = sliding_model();
        let mut authority = WorkbookAuthority::from_model_with_client_id(&model, 61).unwrap();
        let commit = |authority: &mut WorkbookAuthority, to| {
            let ops = [Op::SetChartAnchor {
                sheet: SheetId(0),
                frame: "xl/drawings/drawing1.xml#0".to_owned(),
                part: "xl/charts/chart1.xml".to_owned(),
                from: authority.materialize().unwrap().sheets[0].charts[0].anchor,
                to,
            }];
            let staged = authority
                .stage_local_ops_v1(&ops, SyncOrigin::User, &Stylesheet::default())
                .unwrap();
            authority
                .apply_local_update_v1(&staged.update, SyncOrigin::User)
                .unwrap();
        };

        commit(&mut authority, slid_anchor(2));
        let checkpoint = authority.checkpoint();
        let vector = authority.encode_state_vector_v1();
        let anchor = authority.materialize().unwrap().sheets[0].charts[0].anchor;
        let depth = authority.undo_depth();
        assert!(depth > 0);

        commit(&mut authority, slid_anchor(5));
        assert_ne!(
            authority.materialize().unwrap().sheets[0].charts[0].anchor,
            anchor
        );

        authority.restore(checkpoint).unwrap();
        assert_eq!(
            authority.materialize().unwrap().sheets[0].charts[0].anchor,
            anchor,
            "restore must bring the document back"
        );
        assert_eq!(
            authority.encode_state_vector_v1(),
            vector,
            "restore must wind the document back, not forward"
        );
        assert_eq!(
            authority.undo_depth(),
            depth,
            "restore must bring back the stacks"
        );

        // the history still belongs to this document, so it can still be spent
        let undone = authority
            .undo()
            .expect("a restored history must still apply")
            .expect("the entry is on the stack");
        assert_eq!(
            undone.model.sheets[0].charts[0].anchor,
            model.sheets[0].charts[0].anchor
        );
        assert_eq!(authority.undo_depth(), depth - 1);
    }

    /// A peer's chart write can lose the merge to a concurrent local one and
    /// sit in the document unseen. Undoing the local write must not strand the
    /// authority on state it cannot read: the step is refused and the replica
    /// stays exactly as usable as it was.
    #[test]
    fn a_hidden_chart_conflict_leaves_the_replica_usable() {
        let model = sliding_model();
        let mut authority = WorkbookAuthority::from_model_with_client_id(&model, 304).unwrap();
        let frozen = authority.structure().unwrap();
        let hostile = peer_chart_update(
            &authority,
            111,
            r#"[{"part":"xl/charts/chart1.xml","drawing":"xl/drawings/drawing1.xml","anchorIndex":0,"anchor":{"kind":"twoCell","from":{"col":0,"colOff":0,"row":0,"rowOff":0},"to":{"col":4,"colOff":0,"row":8,"rowOff":0},"edit_as":"twoCell"},"refs":[{"kind":"values","formula":"Hijacked!$A$1"}]}]"#,
        );

        let ops = [Op::SetChartAnchor {
            sheet: SheetId(0),
            frame: "xl/drawings/drawing1.xml#0".to_owned(),
            part: "xl/charts/chart1.xml".to_owned(),
            from: model.sheets[0].charts[0].anchor,
            to: slid_anchor(2),
        }];
        let local = authority
            .stage_local_ops_v1(&ops, SyncOrigin::User, &Stylesheet::default())
            .unwrap();
        authority
            .apply_local_update_v1(&local.update, SyncOrigin::User)
            .unwrap();

        // the hostile value loses the merge, so the freeze sees only the move.
        let staged = authority.stage_updates_v1(&[&hostile], None).unwrap();
        assert_eq!(staged.structure, frozen);
        authority
            .apply_staged_update_v1(&staged.commit_update)
            .unwrap();
        let merged = authority.materialize().unwrap();
        assert_eq!(
            merged.sheets[0].charts[0].refs,
            model.sheets[0].charts[0].refs
        );

        // undoing the move succeeds, takes the anchor back, and never lets the
        // hidden remap through.
        let undone = authority
            .undo()
            .expect("a hidden conflict must not fail the undo")
            .expect("the move is on the stack, so undo must do something");
        assert_eq!(undone.structure, frozen);
        let after = authority.materialize().unwrap();
        assert_eq!(
            after.sheets[0].charts[0].refs,
            model.sheets[0].charts[0].refs
        );
        assert_eq!(
            after.sheets[0].charts[0].anchor, model.sheets[0].charts[0].anchor,
            "undo must put the chart back where it started"
        );
        assert_eq!(authority.structure().unwrap(), frozen);
        assert!(
            !authority.can_undo(),
            "one undo must consume exactly the one entry, not drain a longer stack"
        );
        let _ = merged;
    }
}