blockworx_doc/

session.rs

//! The two ends of the document: [`Host`], the authority that sequences
//! commits, and [`ClientSession`], the edit surface that predicts them.
//! Neither knows about storage — that wraps them.
//! Rationale: `docs/collab-architecture.md` §6-§7.

use std::collections::VecDeque;

use serde::{Deserialize, Serialize};

use crate::{
    block_model::Live,
    commit::Commit,
    document::{Document, FoldError},
    entity::Entity,
    opcode::{Crud, OpCodes},
    rev::{Confirmed, Provisional, Rev, RevKind},
};

/// Matches an answer to the submission that produced it, and names the
/// journal entry that submission left behind.
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug, Serialize, Deserialize)]
pub struct Nonce(u64);

impl Nonce {
    pub(crate) const fn new(count: u64) -> Self {
        Self(count)
    }
}

#[derive(Clone, PartialEq, Eq, Debug, thiserror::Error)]
pub enum SessionError {
    /// The host lost or reordered an answer. Resync.
    #[error("expected rev {expected:?} on this connection, got {got:?}")]
    RevGap { expected: Rev, got: Rev },
    /// The queue and the host disagree about what is in flight.
    #[error("expected an answer for {expected:?}, got one for {got:?}")]
    NonceMismatch { expected: Option<Nonce>, got: Nonce },
    /// A sequenced commit must fold; reaching this means the log or the
    /// host is wrong. Halt and resync, never repair.
    #[error("a sequenced commit was refused: {0}")]
    Corrupt(#[from] FoldError),
}

/// The authority: what `LocalHost` wraps.
#[derive(Default)]
pub struct Host {
    state: Document<Confirmed>,
    log: Vec<Commit>,
}

/// A commit that has folded but is not yet visible. The server persists it
/// before publishing, so a commit it has broadcast always survives a
/// restart — folding first and writing afterwards would let a failed
/// `INSERT` leave clients holding a commit the log does not have.
pub struct Accepted {
    state: Document<Confirmed>,
    commit: Commit,
}

impl Accepted {
    pub fn rev(&self) -> Rev {
        self.state.rev()
    }

    pub fn commit(&self) -> &Commit {
        &self.commit
    }
}

impl Host {
    /// Fold without publishing: validates, mints the rev, and hands back a
    /// successor the caller can persist before making it visible.
    ///
    /// # Errors
    /// The fold's refusal, which consumes no rev and logs nothing.
    pub fn accept(&self, commit: Commit) -> Result<Accepted, FoldError> {
        Ok(Accepted {
            state: self.state.try_apply(&commit)?,
            commit,
        })
    }

    pub fn publish(&mut self, accepted: Accepted) -> Rev {
        self.state = accepted.state;
        self.log.push(accepted.commit);
        self.state.rev()
    }

    /// Accept and publish in one step, for hosts with nothing to persist.
    ///
    /// # Errors
    /// As [`Self::accept`].
    pub fn ingest(&mut self, commit: &Commit) -> Result<Rev, FoldError> {
        let accepted = self.accept(commit.clone())?;
        Ok(self.publish(accepted))
    }

    pub fn state(&self) -> &Document<Confirmed> {
        &self.state
    }

    pub fn rev(&self) -> Rev {
        self.state.rev()
    }

    /// `log[i]` holds `Rev(i + 1)` — contiguous because a refused commit
    /// consumes no rev, which is what lets the log be a `Vec`.
    pub fn commits_after(&self, rev: Rev) -> &[Commit] {
        let already_held = (rev.get() as usize).min(self.log.len());
        &self.log[already_held..]
    }
}

struct Pending {
    nonce: Nonce,
    commit: Commit,
}

/// Which stack a submission's inverse lands on, and what it does to the
/// other — the whole undo/redo policy, stated once. Inverses are built
/// from the values current at use time, not at edit time, which is what
/// keeps an undo/redo round trip from moving the document.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum JournalAs {
    Edit,
    Undo,
    Redo,
}

struct JournalEntry {
    /// The submission this entry inverts, and the entry's identity.
    ///
    /// Kept after the acknowledgement, unlike [`Self::pending`], because the
    /// editor's own stack names its steps by it: a rejection removes an entry
    /// from the *middle* of this stack, so a caller that identified an entry
    /// by position would afterwards invert a different edit than the one it
    /// meant.
    nonce: Nonce,
    inverse: Commit,
    /// Still in flight, so a rejection can still take this entry away.
    pending: bool,
}

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

impl Journal {
    fn push(&mut self, entry: JournalEntry, journal_as: JournalAs) {
        match journal_as {
            JournalAs::Edit => {
                self.redo.clear();
                self.undo.push(entry);
            }
            JournalAs::Undo => self.redo.push(entry),
            JournalAs::Redo => self.undo.push(entry),
        }
    }

    fn ack(&mut self, nonce: Nonce) {
        for entry in self.undo.iter_mut().chain(self.redo.iter_mut()) {
            if entry.nonce == nonce {
                entry.pending = false;
            }
        }
    }

    /// The inverse of an edit that never landed would write stale values
    /// at a fresh order, and a refused create's inverse cannot fold.
    fn discard(&mut self, nonce: Nonce) {
        self.undo
            .retain(|entry| !(entry.pending && entry.nonce == nonce));
        self.redo
            .retain(|entry| !(entry.pending && entry.nonce == nonce));
    }
}

/// Why a history step did not happen.
#[derive(Debug, thiserror::Error, PartialEq)]
pub enum UndoRefusal {
    #[error("nothing stands ready to be inverted")]
    Spent,
    #[error("the step names an edit this session no longer holds on top ({top:?})")]
    Stale { top: Nonce },
    #[error("the inverse was refused: {0}")]
    Refused(#[from] FoldError),
}

/// The edit surface: what the editor wraps.
#[derive(Default)]
pub struct ClientSession {
    confirmed: Document<Confirmed>,
    /// Sealed, submitted, unacked — in the order the server answers in.
    pending: VecDeque<Pending>,
    /// `confirmed ⊕ pending`: what the editor draws.
    optimistic: Document<Provisional>,
    journal: Journal,
    next_nonce: u64,
}

impl ClientSession {
    /// # Errors
    /// [`SessionError::Corrupt`] if the log and this build disagree.
    pub fn welcome(commits: &[Commit]) -> Result<Self, SessionError> {
        let mut session = Self::default();
        for commit in commits {
            session.confirmed = session.confirmed.try_apply(commit)?;
        }
        session.rebuild();
        Ok(session)
    }

    /// The only rev that means anything off this machine.
    pub fn rev(&self) -> Rev {
        self.confirmed.rev()
    }

    pub fn confirmed(&self) -> &Document<Confirmed> {
        &self.confirmed
    }

    pub fn optimistic(&self) -> &Document<Provisional> {
        &self.optimistic
    }

    /// What the transport sends. Undo and redo build their own commits,
    /// so this is the caller's only handle on them.
    pub fn last_submission(&self) -> Option<(Nonce, &Commit)> {
        self.pending
            .back()
            .map(|pending| (pending.nonce, &pending.commit))
    }

    pub fn can_undo(&self) -> bool {
        !self.journal.undo.is_empty()
    }

    /// How many of this client's own edits stand ready to be inverted. The
    /// editor keeps a parallel stack that interleaves view steps with these,
    /// and the two depths must agree — exposed so the mismatch is checkable
    /// rather than silent.
    pub fn undo_depth(&self) -> usize {
        self.journal.undo.len()
    }

    pub fn redo_depth(&self) -> usize {
        self.journal.redo.len()
    }

    pub fn can_redo(&self) -> bool {
        !self.journal.redo.is_empty()
    }

    /// # Errors
    /// The fold's refusal, checked against the prediction: the server
    /// would refuse it too, so it is never queued.
    pub fn submit(&mut self, commit: Commit) -> Result<Nonce, FoldError> {
        self.submit_journaled(commit, JournalAs::Edit)
    }

    fn submit_journaled(
        &mut self,
        commit: Commit,
        journal_as: JournalAs,
    ) -> Result<Nonce, FoldError> {
        // One binding for both reads: the baseline must come from the
        // document the commit is predicted against, never from `confirmed`.
        let pre_image = &self.optimistic;
        let predicted = pre_image.try_apply(&commit)?;
        let inverse = inverse_of(pre_image, &commit);

        let nonce = Nonce::new(self.next_nonce);
        self.next_nonce += 1;
        self.journal.push(
            JournalEntry {
                nonce,
                inverse,
                pending: true,
            },
            journal_as,
        );
        self.pending.push_back(Pending { nonce, commit });
        // Appending to the queue makes this what a rebuild would produce.
        self.optimistic = predicted;
        Ok(nonce)
    }

    /// The edit an [`undo`](Self::undo) would invert, and the one a
    /// [`redo`](Self::redo) would replay — so a caller holding its own
    /// history can tell whether the step it is about to take is still the
    /// step this session would take.
    pub fn next_undo(&self) -> Option<Nonce> {
        self.journal.undo.last().map(|entry| entry.nonce)
    }

    pub fn next_redo(&self) -> Option<Nonce> {
        self.journal.redo.last().map(|entry| entry.nonce)
    }

    /// Every edit standing ready to be inverted, oldest first — for a caller
    /// whose own stack must agree with this one, entry for entry.
    pub fn undo_nonces(&self) -> Vec<Nonce> {
        self.journal.undo.iter().map(|entry| entry.nonce).collect()
    }

    /// Submits the inverse of `edit` as an ordinary commit. The entry stays
    /// on the stack if its inverse will not fold.
    ///
    /// `edit` names which submission the caller believes it is inverting,
    /// and a mismatch is refused rather than silently taken as "the top one".
    /// Identity is the point: a rejected commit is discarded from the middle
    /// of the stack, so an "undo the top" call from a caller whose own
    /// history still counts that edit would invert its neighbour.
    ///
    /// # Errors
    /// [`UndoRefusal::Spent`] when nothing stands ready, [`UndoRefusal::Stale`]
    /// when the top is a different edit than `edit` — its own nonce reported
    /// so the caller can resynchronise — and [`UndoRefusal::Refused`] for a
    /// fold refusal, as [`Self::submit`].
    pub fn undo(&mut self, edit: Nonce) -> Result<Nonce, UndoRefusal> {
        self.step(edit, JournalAs::Undo)
    }

    /// # Errors
    /// As [`Self::undo`].
    pub fn redo(&mut self, edit: Nonce) -> Result<Nonce, UndoRefusal> {
        self.step(edit, JournalAs::Redo)
    }

    fn step(&mut self, edit: Nonce, direction: JournalAs) -> Result<Nonce, UndoRefusal> {
        let stack = match direction {
            JournalAs::Undo => &mut self.journal.undo,
            JournalAs::Redo => &mut self.journal.redo,
            JournalAs::Edit => unreachable!("an edit is not a history step"),
        };
        let top = stack.last().ok_or(UndoRefusal::Spent)?;
        if top.nonce != edit {
            return Err(UndoRefusal::Stale { top: top.nonce });
        }
        let inverse = top.inverse.clone();
        let submitted = self.submit_journaled(inverse, direction)?;
        match direction {
            JournalAs::Undo => self.journal.undo.pop(),
            JournalAs::Redo => self.journal.redo.pop(),
            JournalAs::Edit => None,
        };
        Ok(submitted)
    }

    /// Someone else's edit.
    ///
    /// # Errors
    /// [`SessionError::RevGap`] out of sequence, [`SessionError::Corrupt`]
    /// if a sequenced commit refuses.
    pub fn apply_foreign(&mut self, rev: Rev, commit: &Commit) -> Result<(), SessionError> {
        self.expect_next(rev)?;
        self.confirmed = self.confirmed.try_apply(commit)?;
        self.rebuild();
        Ok(())
    }

    /// Our own edit, accepted and sequenced as `rev`.
    ///
    /// # Errors
    /// [`SessionError::NonceMismatch`] if the answer is not for the front
    /// of the queue, plus [`Self::apply_foreign`]'s.
    pub fn committed(&mut self, nonce: Nonce, rev: Rev) -> Result<(), SessionError> {
        self.expect_front(nonce)?;
        self.expect_next(rev)?;
        if let Some(acked) = self.pending.front() {
            self.confirmed = self.confirmed.try_apply(&acked.commit)?;
            self.pending.pop_front();
            self.journal.ack(nonce);
            self.rebuild();
        }
        Ok(())
    }

    /// Our own edit, refused at ingress: nothing was sequenced.
    ///
    /// # Errors
    /// [`SessionError::NonceMismatch`], as [`Self::committed`].
    pub fn rejected(&mut self, nonce: Nonce) -> Result<(), SessionError> {
        self.expect_front(nonce)?;
        self.pending.pop_front();
        self.journal.discard(nonce);
        self.rebuild();
        Ok(())
    }

    fn expect_next(&self, rev: Rev) -> Result<(), SessionError> {
        let expected = self.confirmed.rev().next();
        debug_assert_eq!(rev, expected, "a gap in the per-connection rev sequence");
        if rev == expected {
            Ok(())
        } else {
            Err(SessionError::RevGap { expected, got: rev })
        }
    }

    fn expect_front(&self, nonce: Nonce) -> Result<(), SessionError> {
        let front = self.pending.front().map(|pending| pending.nonce);
        if front == Some(nonce) {
            Ok(())
        } else {
            Err(SessionError::NonceMismatch {
                expected: front,
                got: nonce,
            })
        }
    }

    /// `confirmed ⊕ pending`, from scratch. Each fold mints its own
    /// provisional rev, so there is no rev arithmetic here.
    fn rebuild(&mut self) {
        self.optimistic = self
            .pending
            .iter()
            .fold(self.confirmed.predict(), |doc, pending| {
                match doc.try_apply(&pending.commit) {
                    Ok(predicted) => predicted,
                    // Skipped, not dropped: a foreign commit can invalidate
                    // a pending one, and only the server may retire a queue
                    // entry. The edit reappears if a later commit revalidates
                    // it.
                    Err(_) => doc,
                }
            });
    }
}

/// The commit that undoes `commit`, read against the document as it stood
/// *before* it, ops walked forward and the list reversed.
///
/// The reversal is not about repeated writes to one register — those carry
/// the same pre-image value either way round. It is about lifecycle
/// *pairs*: `[Create, Delete]` inverts to `[Delete, Restore]`, which run
/// forwards would leave alive an entity that never existed.
fn inverse_of<R: RevKind>(pre_image: &Document<R>, commit: &Commit) -> Commit {
    let mut ops: Vec<OpCodes> = commit
        .ops()
        .iter()
        .filter_map(|op| invert_op(pre_image, op))
        .collect();
    ops.reverse();
    Commit::new(format!("Undo {}", commit.label()), ops)
}

fn invert_op<R: RevKind>(pre_image: &Document<R>, op: &OpCodes) -> Option<OpCodes> {
    match op {
        OpCodes::Document(update) => {
            Some(OpCodes::Document(pre_image.title_block().invert(update)))
        }
        OpCodes::Block(id, crud) => {
            invert_crud(pre_image.block(id), crud).map(|crud| OpCodes::Block(*id, crud))
        }
        OpCodes::Pin(id, crud) => {
            invert_crud(pre_image.pin(id), crud).map(|crud| OpCodes::Pin(*id, crud))
        }
        OpCodes::Route(id, crud) => {
            invert_crud(pre_image.route(id), crud).map(|crud| OpCodes::Route(*id, crud))
        }
        OpCodes::RouteLabel(id, crud) => {
            invert_crud(pre_image.route_label(id), crud).map(|crud| OpCodes::RouteLabel(*id, crud))
        }
        OpCodes::Text(id, crud) => {
            invert_crud(pre_image.text(id), crud).map(|crud| OpCodes::Text(*id, crud))
        }
        OpCodes::Comment(id, crud) => {
            invert_crud(pre_image.comment(id), crud).map(|crud| OpCodes::Comment(*id, crud))
        }
        OpCodes::Image(id, crud) => {
            invert_crud(pre_image.image(id), crud).map(|crud| OpCodes::Image(*id, crud))
        }
        // An asset payload has no inverse to build: it is create-only, so
        // there is no un-create op to emit, and the bytes are meant to
        // outlive the reference that brought them in — undoing the
        // placement leaves an unreferenced payload for compaction, not a
        // dangling reference for the next redo.
        OpCodes::Asset(..) => None,
    }
}

/// `None` drops the op from the inverse: an update to an entity the same
/// commit created has nothing to restore, and the create's `Delete` —
/// which runs after it once the list is reversed — removes it wholesale.
/// Presence is map presence, not liveness: the tombstone retains the
/// inner, which is what makes the lifecycle arms baseline-free.
fn invert_crud<E: Entity>(
    pre_image: Option<&Live<E>>,
    crud: &Crud<E::Init, E::Update>,
) -> Option<Crud<E::Init, E::Update>> {
    match crud {
        Crud::Update(update) => Some(Crud::Update(pre_image?.as_ref().invert(update))),
        lifecycle => lifecycle.invert_lifecycle(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fixtures::{block_id, commit, image_id, pin_id, projection};
    use crate::{
        block_model::{Asset, BlockInit, BlockUpdate, Icon, ImageInit, LabelInit, PinInit},
        document::TitleBlockUpdate,
        geometry::{FracVal, GridPoint, GridRect, GridSize, PinSlot, ScreenRect},
        id::BlockId,
        values::{LabelSide, PinDir, Role},
    };

    fn rect(x: i32, y: i32) -> GridRect {
        GridRect {
            top_left: GridPoint { x, y },
            size: GridSize { w: 4, h: 4 },
        }
    }

    fn label_init(name: String) -> LabelInit {
        LabelInit {
            name,
            side: LabelSide::default(),
            offset: FracVal::default(),
            hidden: false,
        }
    }

    fn block_create(byte: u8) -> OpCodes {
        OpCodes::Block(
            block_id(byte),
            Crud::Create(BlockInit {
                parent: BlockId::NULL,
                rect: rect(0, 0),
                locked: false,
                role: Role::default(),
                title: label_init(format!("b{byte}")),
                type_label: label_init(String::new()),
                icon: Icon::default(),
            }),
        )
    }

    fn reparent(child: u8, parent: BlockId) -> OpCodes {
        OpCodes::Block(block_id(child), Crud::Update(BlockUpdate::Parent(parent)))
    }

    fn resize(byte: u8, to: GridRect) -> OpCodes {
        OpCodes::Block(block_id(byte), Crud::Update(BlockUpdate::Rect(to)))
    }

    /// A pin whose owner block does not exist — refused by the fold's
    /// commit-end validation, on either end of the wire.
    fn orphan_pin(byte: u8) -> OpCodes {
        OpCodes::Pin(
            pin_id(byte),
            Crud::Create(PinInit {
                owner: block_id(200),
                name: format!("p{byte}"),
                type_name: String::new(),
                tag: String::new(),
                tag_hidden: false,
                rect: rect(0, 0),
                slot: PinSlot::default(),
                dir: PinDir::default(),
                port_accent: Role::default(),
                flip_lr: false,
            }),
        )
    }

    fn parent_of(session: &ClientSession, byte: u8) -> BlockId {
        *session
            .optimistic()
            .block(&block_id(byte))
            .expect("the block is in the document")
            .as_ref()
            .parent
            .as_ref()
    }

    /// Sequence the session's last submission at the host and ack it —
    /// one round trip with nothing else in flight.
    fn ack_last(host: &mut Host, session: &mut ClientSession) {
        let (nonce, commit) = session.last_submission().expect("a submission is queued");
        let commit = commit.clone();
        let rev = host.ingest(&commit).expect("the host accepts it");
        session.committed(nonce, rev).expect("the ack lands");
    }

    fn edit(host: &mut Host, session: &mut ClientSession, commit: Commit) {
        session.submit(commit).expect("the edit folds");
        ack_last(host, session);
    }

    /// A host and a session agreeing on one top-level block.
    fn one_block() -> (Host, ClientSession) {
        let (mut host, mut session) = (Host::default(), ClientSession::default());
        edit(
            &mut host,
            &mut session,
            commit("Added a block", vec![block_create(1)]),
        );
        (host, session)
    }

    #[test]
    fn an_ack_leaves_the_prediction_equal_to_the_confirmed_document() {
        let (host, session) = one_block();

        assert!(session.last_submission().is_none(), "the queue drains");
        assert_eq!(
            session.optimistic().content_hash(),
            session.confirmed().content_hash(),
            "with nothing pending the prediction is the confirmed document",
        );
        assert_eq!(
            session.confirmed().content_hash(),
            host.state().content_hash(),
            "and matches the host's fold, revs included",
        );
    }

    /// The reconciliation oracle in miniature: while an edit is in flight,
    /// the prediction is exactly the confirmed head with the queue folded
    /// on top.
    #[test]
    fn a_foreign_commit_rebuilds_the_prediction_over_the_new_head() {
        let (mut host, mut session) = one_block();
        session
            .submit(commit("Moved it", vec![resize(1, rect(5, 5))]))
            .expect("the edit folds");

        let foreign = commit("Added another", vec![block_create(2)]);
        let rev = host.ingest(&foreign).expect("the host accepts it");
        session
            .apply_foreign(rev, &foreign)
            .expect("the delivery lands");

        let pending = session
            .last_submission()
            .expect("the edit is still in flight")
            .1
            .clone();
        let from_scratch = session
            .confirmed()
            .try_apply(&pending)
            .expect("the pending edit folds on the new head");
        assert_eq!(
            session.optimistic().content_hash(),
            from_scratch.content_hash(),
            "optimistic == confirmed ⊕ pending, byte for byte",
        );
        assert_eq!(
            *session
                .optimistic()
                .block(&block_id(1))
                .expect("block 1 is present")
                .as_ref()
                .rect
                .as_ref(),
            rect(5, 5),
            "the unacked local edit still stands above the foreign one",
        );
    }

    #[test]
    #[should_panic(expected = "gap in the per-connection rev sequence")]
    fn a_rev_gap_panics_in_debug() {
        let (_host, mut session) = one_block();
        let skipped = session.rev().next().next();
        let _ = session.apply_foreign(skipped, &commit("Skipped ahead", vec![block_create(9)]));
    }

    #[test]
    fn an_answer_for_anything_but_the_front_of_the_queue_is_refused() {
        let (host, mut session) = one_block();
        let first = session
            .submit(commit("First", vec![resize(1, rect(1, 1))]))
            .expect("the edit folds");
        let second = session
            .submit(commit("Second", vec![resize(1, rect(2, 2))]))
            .expect("the edit folds");
        assert_ne!(first, second, "each submission gets its own nonce");

        assert_eq!(
            session.committed(second, host.rev().next()),
            Err(SessionError::NonceMismatch {
                expected: Some(first),
                got: second,
            }),
            "answers arrive in submission order",
        );
    }

    #[test]
    fn a_local_edit_that_cannot_fold_is_never_queued() {
        let mut session = ClientSession::default();
        let before = session.optimistic().content_hash();

        assert!(
            session
                .submit(commit("Orphan pin", vec![orphan_pin(3)]))
                .is_err(),
            "the client refuses what the server would refuse",
        );
        assert!(session.last_submission().is_none(), "nothing was queued");
        assert!(!session.can_undo(), "and nothing was journalled");
        assert_eq!(
            session.optimistic().content_hash(),
            before,
            "the prediction is untouched",
        );
    }

    /// The case identity exists for. A rejection discards its entry from the
    /// *middle* of the stack, so after one lands the stack's positions no
    /// longer mean what a caller counting its own steps thinks they mean.
    ///
    /// A caller that says which edit it means gets told; one that said "the
    /// top" would silently invert its neighbour.
    #[test]
    fn a_rejection_in_the_middle_makes_a_step_stale_rather_than_wrong() {
        let (mut host, mut session) = one_block();
        // Three edits: the first acknowledged, the next two still in flight.
        edit(
            &mut host,
            &mut session,
            commit("Moved it", vec![resize(1, rect(5, 5))]),
        );
        let first = session
            .next_undo()
            .expect("the acknowledged edit is journalled");
        let doomed = session
            .submit(commit("Moved it again", vec![resize(1, rect(6, 6))]))
            .expect("the edit folds locally");
        let last = session
            .submit(commit("And again", vec![resize(1, rect(7, 7))]))
            .expect("the edit folds locally");
        assert!(
            session.undo_nonces().ends_with(&[first, doomed, last]),
            "precondition: the doomed edit sits in the middle, not on top",
        );

        session.rejected(doomed).expect("the rejection lands");

        let after = session.undo_nonces();
        assert!(
            after.ends_with(&[first, last]),
            "the rejected entry left from the middle, not from the top: {after:?}",
        );
        assert!(!after.contains(&doomed));
        assert_eq!(
            session.undo(first),
            Err(UndoRefusal::Stale { top: last }),
            "a step naming the wrong edit is refused, not taken as the top",
        );
        assert!(
            session.undo(last).is_ok(),
            "and the step that names the top is taken",
        );
    }

    /// A spent stack refuses rather than reporting a step it did not take.
    #[test]
    fn a_step_on_a_spent_stack_is_refused() {
        let (_host, mut session) = one_block();
        let only = session.next_undo().expect("the setup edit is journalled");
        session.undo(only).expect("the inverse folds");
        assert_eq!(session.next_undo(), None);
        assert_eq!(session.undo(only), Err(UndoRefusal::Spent));
    }

    #[test]
    fn a_rejection_drops_the_edit_and_its_journal_entry() {
        let (mut host, mut session) = one_block();
        let confirmed = session.confirmed().content_hash();

        let doomed = session
            .submit(commit("Moved it", vec![resize(1, rect(5, 5))]))
            .expect("the edit folds locally");
        session.rejected(doomed).expect("the rejection lands");

        assert_eq!(
            session.optimistic().content_hash(),
            confirmed,
            "the edit visibly reverts",
        );
        assert!(session.can_undo(), "the accepted create is still undoable");

        // The surviving entry must be the create's, not the rejected move's:
        // undoing an edit that never landed would write stale values at a
        // fresh order.
        let top = session.next_undo().expect("something to undo");
        session.undo(top).expect("the inverse folds");
        ack_last(&mut host, &mut session);
        assert!(
            projection(session.confirmed()).is_empty(),
            "undo removed the block, so the entry inverted the create",
        );
    }

    /// The decision this step turns on: a foreign commit can invalidate a
    /// pending one, and the queue answers to the server rather than to
    /// this frame's paint.
    #[test]
    fn a_pending_commit_the_head_invalidated_is_skipped_then_returns() {
        let (mut host, mut session) = (Host::default(), ClientSession::default());
        let setup = commit("Two blocks", vec![block_create(1), block_create(2)]);
        let rev = host.ingest(&setup).expect("the host accepts it");
        session.apply_foreign(rev, &setup).expect("the setup lands");

        let nonce = session
            .submit(commit("Nested 1 under 2", vec![reparent(1, block_id(2))]))
            .expect("the edit folds on today's head");
        assert_eq!(parent_of(&session, 1), block_id(2), "predicted immediately");

        let cycle = commit("Nested 2 under 1", vec![reparent(2, block_id(1))]);
        let rev = host.ingest(&cycle).expect("no cycle at the host yet");
        session
            .apply_foreign(rev, &cycle)
            .expect("the delivery lands");
        assert_eq!(
            parent_of(&session, 1),
            BlockId::NULL,
            "the pending edit would close a cycle, so it is not predicted",
        );

        let undo_cycle = commit("Freed 2", vec![reparent(2, BlockId::NULL)]);
        let rev = host.ingest(&undo_cycle).expect("the host accepts it");
        session
            .apply_foreign(rev, &undo_cycle)
            .expect("the delivery lands");
        assert_eq!(
            parent_of(&session, 1),
            block_id(2),
            "and reappears once the head makes it valid again",
        );

        let pending = session
            .last_submission()
            .expect("the edit stayed queued through both rebuilds")
            .1
            .clone();
        let rev = host.ingest(&pending).expect("the host accepts it now");
        session.committed(nonce, rev).expect("the ack lands");
        assert_eq!(
            *session
                .confirmed()
                .block(&block_id(1))
                .expect("block 1 is present")
                .as_ref()
                .parent
                .as_ref(),
            block_id(2),
        );
    }

    #[test]
    fn a_refused_ingest_consumes_no_rev_and_leaves_the_log_alone() {
        let mut host = Host::default();
        let accepted = host
            .ingest(&commit("Added a block", vec![block_create(1)]))
            .expect("the host accepts it");

        assert!(
            host.ingest(&commit("Orphan pin", vec![orphan_pin(3)]))
                .is_err(),
        );
        assert_eq!(host.rev(), accepted, "a refusal mints no rev");
        assert_eq!(host.commits_after(Rev::ZERO).len(), 1, "and logs nothing");

        let next = host
            .ingest(&commit("Added another", vec![block_create(2)]))
            .expect("the host accepts it");
        assert_eq!(
            next,
            accepted.next(),
            "the next commit takes the rev the refusal did not consume",
        );
    }

    #[test]
    fn commits_after_replays_the_tail_in_rev_order() {
        let mut host = Host::default();
        for byte in 1..=3 {
            host.ingest(&commit("Added a block", vec![block_create(byte)]))
                .expect("the host accepts it");
        }

        assert_eq!(host.commits_after(Rev::ZERO).len(), 3, "the whole log");
        assert_eq!(
            host.commits_after(Rev::new(2)).len(),
            1,
            "log[i] is Rev(i+1)"
        );
        assert!(host.commits_after(host.rev()).is_empty(), "caught up");
        assert!(
            host.commits_after(host.rev().next()).is_empty(),
            "and a rev beyond the head does not panic",
        );
    }

    /// `fold(fold(s, C), inverse(C)) == s` on the visible projection, with
    /// the edit asserted observable so the property cannot pass vacuously.
    fn assert_round_trip<R: RevKind>(start: &Document<R>, edit: &Commit) -> Commit {
        let after = start.try_apply(edit).expect("the edit folds");
        assert_ne!(
            projection(&after),
            projection(start),
            "the edit must be observable or the round trip proves nothing",
        );

        let inverse = inverse_of(start, edit);
        let back = after.try_apply(&inverse).expect("the inverse folds");
        assert_eq!(projection(&back), projection(start));
        inverse
    }

    #[test]
    fn the_journal_inverts_a_commit_that_writes_one_register_twice() {
        let start = Document::<Confirmed>::default()
            .try_apply(&commit("Added a block", vec![block_create(1)]))
            .expect("the fold succeeds");

        // Reversal is the point: both inverses carry the pre-image value,
        // so whichever lands last by seq restores it.
        let inverse = assert_round_trip(
            &start,
            &commit(
                "Dragged it",
                vec![resize(1, rect(5, 5)), resize(1, rect(9, 9))],
            ),
        );
        assert_eq!(inverse.ops().len(), 2);
    }

    #[test]
    fn the_journal_inverts_a_create_and_a_delete() {
        let start = Document::<Confirmed>::default()
            .try_apply(&commit("Added a block", vec![block_create(1)]))
            .expect("the fold succeeds");

        assert_round_trip(&start, &commit("Added another", vec![block_create(2)]));
        assert_round_trip(
            &start,
            &commit(
                "Deleted it",
                vec![OpCodes::Block(block_id(1), Crud::Delete)],
            ),
        );
    }

    /// What the reversal is actually for. Two writes to one register are
    /// order-independent (both inverses carry the same pre-image value),
    /// but a lifecycle *pair* is not: `[Create, Delete]` inverts to
    /// `[Delete, Restore]`, which run forwards leaves the entity alive —
    /// resurrecting something that never existed before the commit.
    #[test]
    fn the_inverse_of_a_lifecycle_pair_unwinds_in_reverse() {
        let start = Document::<Confirmed>::default();
        let edit = commit(
            "Added and dropped a block",
            vec![block_create(1), OpCodes::Block(block_id(1), Crud::Delete)],
        );
        let after = start.try_apply(&edit).expect("the edit folds");
        assert!(
            projection(&after).is_empty(),
            "the block ends tombstoned, so it was never visible",
        );

        let inverse = inverse_of(&start, &edit);
        assert!(
            matches!(inverse.ops().last(), Some(OpCodes::Block(_, Crud::Delete))),
            "the create's inverse must run last",
        );
        let back = after.try_apply(&inverse).expect("the inverse folds");
        assert!(
            back.block(&block_id(1)).is_none_or(|live| !live.is_alive()),
            "undo must not resurrect what the commit only ever tombstoned",
        );
    }

    #[test]
    fn the_journal_skips_an_update_to_an_entity_the_commit_created() {
        let start = Document::<Confirmed>::default();
        let inverse = assert_round_trip(
            &start,
            &commit(
                "Added and placed a block",
                vec![block_create(1), resize(1, rect(5, 5))],
            ),
        );

        assert_eq!(
            inverse.ops().len(),
            1,
            "the update has no baseline to restore; the create's delete covers it",
        );
        assert!(matches!(
            inverse.ops().first(),
            Some(OpCodes::Block(_, Crud::Delete)),
        ));
    }

    /// A payload op has no inverse — undoing the placement un-creates the
    /// image and leaves the bytes behind, unreferenced. Emitting a
    /// "delete the asset" here is what create-only rules out: a redo, or
    /// anyone else's placement of the same file, still resolves.
    #[test]
    fn undoing_a_placement_un_creates_the_image_and_keeps_its_payload() {
        let asset = Asset::Png(b"artwork".as_slice().into());
        let start = Document::<Confirmed>::default()
            .try_apply(&commit("Added a block", vec![block_create(1)]))
            .expect("the fold succeeds");
        let edit = commit(
            "Placed an image",
            vec![
                OpCodes::Asset(asset.hash(), asset.clone()),
                OpCodes::Image(
                    image_id(2),
                    Crud::Create(ImageInit {
                        owner: block_id(1),
                        asset: asset.hash(),
                        rect: ScreenRect::default(),
                    }),
                ),
            ],
        );

        let after = start.try_apply(&edit).expect("the edit folds");
        assert!(
            after.image(&image_id(2)).is_some_and(Live::is_alive),
            "precondition: the placement landed",
        );
        assert_eq!(after.asset(&asset.hash()), Some(&asset));

        let inverse = inverse_of(&start, &edit);
        assert_eq!(
            inverse.ops().len(),
            1,
            "only the image create inverts; the payload has no inverse to emit",
        );
        let back = after.try_apply(&inverse).expect("the inverse folds");
        assert!(
            back.image(&image_id(2)).is_none_or(|live| !live.is_alive()),
            "the placement is undone",
        );
        assert_eq!(
            back.asset(&asset.hash()),
            Some(&asset),
            "and the bytes stay: create-only, with reclamation left to compaction",
        );
    }

    #[test]
    fn the_journal_inverts_the_singleton() {
        let start = Document::<Confirmed>::default()
            .try_apply(&commit(
                "Named it",
                vec![OpCodes::Document(TitleBlockUpdate::Name("first".into()))],
            ))
            .expect("the fold succeeds");

        let edit = commit(
            "Renamed it",
            vec![OpCodes::Document(TitleBlockUpdate::Name("second".into()))],
        );
        let after = start.try_apply(&edit).expect("the edit folds");
        let back = after
            .try_apply(&inverse_of(&start, &edit))
            .expect("the inverse folds");

        assert_eq!(after.title_block().name.as_ref().as_str(), "second");
        assert_eq!(back.title_block().name.as_ref().as_str(), "first");
    }

    #[test]
    fn undo_and_redo_travel_as_ordinary_commits() {
        let (mut host, mut session) = one_block();
        let placed = projection(session.confirmed());
        edit(
            &mut host,
            &mut session,
            commit("Moved it", vec![resize(1, rect(5, 5))]),
        );
        let moved = projection(session.confirmed());
        assert_ne!(placed, moved);

        let top = session.next_undo().expect("something to undo");
        session.undo(top).expect("the inverse folds");
        ack_last(&mut host, &mut session);
        assert_eq!(projection(session.confirmed()), placed, "undo restored it");
        assert!(session.can_redo());

        let top = session.next_redo().expect("something to redo");
        session.redo(top).expect("the inverse folds");
        ack_last(&mut host, &mut session);
        assert_eq!(projection(session.confirmed()), moved, "redo re-applied it");
        assert!(session.can_undo());
        assert!(!session.can_redo(), "the redo future is spent");
    }

    #[test]
    fn a_fresh_edit_abandons_the_redo_future() {
        let (mut host, mut session) = one_block();
        edit(
            &mut host,
            &mut session,
            commit("Moved it", vec![resize(1, rect(5, 5))]),
        );
        let top = session.next_undo().expect("something to undo");
        session.undo(top).expect("the inverse folds");
        ack_last(&mut host, &mut session);
        assert!(session.can_redo());

        edit(
            &mut host,
            &mut session,
            commit("Moved it elsewhere", vec![resize(1, rect(7, 7))]),
        );
        assert!(!session.can_redo(), "a new edit forks the history");
    }

    #[test]
    fn welcome_folds_a_log_into_a_session() {
        let mut host = Host::default();
        for byte in 1..=3 {
            host.ingest(&commit("Added a block", vec![block_create(byte)]))
                .expect("the host accepts it");
        }

        let session = ClientSession::welcome(host.commits_after(Rev::ZERO)).expect("the log folds");
        assert_eq!(session.rev(), host.rev());
        assert_eq!(
            session.confirmed().content_hash(),
            host.state().content_hash(),
            "a replayed log reproduces the host's document exactly",
        );
    }
}