lgwks_bot 2.2.0

//! Durable journal: the append that has to land before the irreversible
//! boundary.
//!
//! [`crate::effect`] supplies the identity a settlement is about. This module
//! supplies the place that identity is written down *before* anything
//! irreversible happens, because an identity held only in memory is lost at the
//! exact moment it is needed. A controller that hands bytes to an external
//! system and then crashes has no way to say whether the bytes arrived, and a
//! controller that guesses will either duplicate a non-idempotent effect or
//! silently drop one.
//!
//! # The promise a durability profile makes
//!
//! [`DurabilityPromise`] is graded, and the grade is the whole point: an
//! in-memory adapter can promise nothing across a process exit, so it reports
//! [`DurabilityPromise::Ephemeral`] and [`EffectJournal::admit_external_handoff`]
//! refuses it. That refusal is the load-bearing part of this module. A journal
//! that accepts an append and loses it on restart is worse than no journal,
//! because it converts "I do not know" into a false "I never sent it".
//!
//! Nothing here can verify a promise. A filesystem, a device cache or a
//! virtualization layer can each accept a write and lose it anyway, so a
//! [`DurableAck`] records the promise that was claimed rather than implying one
//! was proven. The tests that matter for that claim are crash tests against a
//! real backing store, and they are not unit tests.
//!
//! # Why compare-and-append
//!
//! Two controllers recovering the same run must not both dispatch. `&mut self`
//! fences within one process, which is not enough: the failure this guards is a
//! second process. So every append states the tail it believes is committed, and
//! an append whose belief is stale is refused rather than merged. A refused
//! append has changed nothing, which is what makes it safe to recover from.
//!
//! [`EffectKey`]: crate::effect::EffectKey
//! [`DurableAck`]: crate::journal::DurableAck
//! [`DurabilityPromise`]: crate::journal::DurabilityPromise
//! [`EffectJournal`]: crate::journal::EffectJournal
//! [`DurabilityPromise::Ephemeral`]: crate::journal::DurabilityPromise::Ephemeral
//! [`EffectJournal::admit_external_handoff`]: crate::journal::EffectJournal::admit_external_handoff

use core::fmt;
use std::collections::HashMap;
use std::io;

use lgwks_std::hash::{Digest, Hasher, blake3};
use lgwks_std::wire::{AlignedVec, WireError};

use crate::effect::{EffectKey, Id128};
use frame::SaturatingFrom;

/// The evidence vocabulary is `crate::ecs`'s, re-exported here rather than
/// restated.
///
/// The journal and the ledger answer the same question, "what does the evidence
/// say about whether the effect landed", and a second enum with the same two
/// arms would drift from this one the first time an arm was added. Re-exporting
/// keeps one definition with two consumers.
pub use crate::ecs::EffectEvidence;

use crate::BoxFuture;

mod wire_form;

/// The two journal records whose archived form is an enum, re-exported from the
/// private `wire_form` module, which records why they are declared there.
pub use wire_form::{
    ArchivedEffectEvent, ArchivedVerificationResult, EffectEvent, VerificationResult,
};

mod file;

/// The frame grammar the file journal and the run store share.
///
/// Private to the crate because a caller cannot frame a record for either store:
/// what a head is over, and what a frame even holds, is each store's own. The
/// grammar — length prefix, torn-tail rule, impossible-length refusal — is one
/// definition both call, so the two stores cannot disagree about what a torn tail
/// is.
pub(crate) mod frame;

/// The one thread that owns a file-backed store's device.
///
/// Extracted from `file` because the run store inherited the same blocking-write
/// defect and needed the same mechanism, and a second owner thread beside the
/// first would be two poison latches with two behaviours.
pub(crate) mod owner;

/// Continue-as-new: a sealed checkpoint and the successor opened from it.
///
/// Re-exported beside the two adapters because it is what a caller asks *of* an
/// adapter rather than what an adapter is: the policy lives here, the file that
/// can carry it out is `FileJournal`, and the trait's two continuation methods
/// are what joins them.
pub mod continuation;

pub use continuation::{
    CONTINUATION_WATERMARK_DENOMINATOR, CONTINUATION_WATERMARK_NUMERATOR, Continuation,
    ContinuationPolicy, ContinuationWatermark, MAX_CHECKPOINT_SETTLED, MAX_CHECKPOINT_UNRESOLVED,
    SealPause, SettledAttempt, UnresolvedAttempt, is_generation_path, successor_path,
};

/// The file-backed journal, re-exported from the private `file` module beside
/// the in-memory one: the trait's second shipped adapter, and the one whose
/// promises a process kill can check.
///
/// `StorageGate` is deliberately absent from this list: one `journal::owner`
/// serves both file-backed stores, so the gate is re-exported from its owning
/// module below rather than from either store, and naming it here would make a
/// caller believe the journal owns a type the run store shares.
pub use file::{Corruption, CorruptionKind, FileJournal, Replay};

/// The handle that releases a stalled store's parked flush.
///
/// Public because "what does this bot do while its disk has stopped answering" is
/// a question an operator has to be able to ask on a real process: the caller
/// awaiting the append holds the journal's borrow for the whole wait, so it cannot
/// reach the journal to release its own device.
pub use owner::StorageGate;

/// The domain separator hashed into the genesis position.
///
/// A chain has to start somewhere, and "started from 32 zero bytes" is a value
/// any other hash could coincide with. Separating the domain means the genesis
/// head cannot be mistaken for the hash of an empty or zeroed event.
const GENESIS_DOMAIN: &[u8] = b"lgwks.journal.v1.genesis";

/// Hard ceiling on events retained by either shipped journal adapter.
///
/// The full history remains available; once this count is reached, new events
/// are refused rather than deleting facts needed to resolve an effect or
/// deduplicate a retry. The journal path is full until its owner performs an
/// explicitly safe handoff to a distinct run and journal; this adapter does not
/// compact or rotate it automatically.
pub const MAX_JOURNAL_EVENTS: usize = 100_000;

/// Hard ceiling on encoded frame bytes retained by either shipped adapter.
///
/// The memory adapter applies the same logical-frame accounting even though it
/// retains entries and indexes rather than framed byte buffers.
pub const MAX_JOURNAL_BYTES: u64 = 64 * 1024 * 1024;

/// What a journal can promise about an append that it has acknowledged.
///
/// Ordered from weakest to strongest, and compared by that order: a caller that
/// needs process-crash survival is satisfied by a power-loss promise, never the
/// reverse.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[non_exhaustive]
pub enum DurabilityPromise {
    /// Survives neither a process exit nor a crash. An in-memory adapter
    /// reports this, and this is the only honest thing it can report.
    Ephemeral,
    /// Survives the writing process dying, including a kill. May still be lost
    /// if the machine loses power before the store flushes.
    ProcessCrash,
    /// Survives power loss, to the extent the device honours its own flush
    /// contract.
    PowerLoss,
}

impl DurabilityPromise {
    /// Whether an append under this promise survives the writer dying.
    ///
    /// This is the threshold an external handoff requires. Below it, a
    /// recovered controller cannot distinguish "the effect never left" from
    /// "the record of it leaving was lost", and that is the state in which a
    /// blind resend happens.
    #[must_use]
    pub const fn survives_process_crash(self) -> bool {
        matches!(self, Self::ProcessCrash | Self::PowerLoss)
    }

    /// Whether this promise is at least as strong as `required`.
    ///
    /// A per-append acknowledgment is checked against the promise the handoff
    /// needs, not against the journal's advertisement: a store that claims
    /// `ProcessCrash` and acks `Ephemeral` is weaker than it says, and the
    /// weaker fact is the one that matters (issue #100).
    #[must_use]
    pub const fn meets(self, required: Self) -> bool {
        match (self, required) {
            (_, Self::Ephemeral) => true,
            (Self::ProcessCrash | Self::PowerLoss, Self::ProcessCrash) => true,
            (Self::PowerLoss, Self::PowerLoss) => true,
            (Self::Ephemeral, _) => false,
            (Self::ProcessCrash, Self::PowerLoss) => false,
        }
    }

    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Ephemeral => "ephemeral",
            Self::ProcessCrash => "process_crash",
            Self::PowerLoss => "power_loss",
        }
    }
}

impl fmt::Display for DurabilityPromise {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// Where a committed event sits in one run's journal.
///
/// The pair of sequence and head is the whole point. A sequence alone says
/// which append this was and nothing about what it contained, so two journals
/// that diverged would agree on their sequence numbers. The head is a hash over
/// every event up to and including this one, so a position is a commitment to
/// the entire history behind it.
#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    Hash,
    lgwks_std::wire::Archive,
    lgwks_std::wire::Serialize,
    lgwks_std::wire::Deserialize,
)]
#[rkyv(crate = lgwks_std::wire::rkyv, compare(PartialEq), derive(Debug))]
pub struct JournalPosition {
    /// How many events are committed at or before this position. Zero is the
    /// genesis, which holds no events.
    sequence: u64,
    /// Hash over every committed event, in order, up to and including this one.
    head: Digest,
}

impl JournalPosition {
    /// The position before the first append.
    ///
    /// Its head is a domain-separated constant rather than a zero digest, so a
    /// journal that was never written cannot be confused with one whose first
    /// event hashed to zero.
    #[must_use]
    pub fn genesis() -> Self {
        Self {
            sequence: 0,
            head: blake3(GENESIS_DOMAIN),
        }
    }

    /// How many events are committed at or before this position.
    #[must_use]
    pub const fn sequence(self) -> u64 {
        self.sequence
    }

    /// The hash over every committed event up to and including this one.
    #[must_use]
    pub const fn head(self) -> Digest {
        self.head
    }
}

impl fmt::Display for JournalPosition {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{} @ {}", self.sequence, self.head)
    }
}

/// Which kind of fact an [`EffectEvent`] records.
///
/// Separate from the event because the ordering ladder is stated in kinds, and
/// a refusal that names the two kinds involved reads in a log line in a way
/// that a rendered event carrying its full key does not.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum EventKind {
    /// The intent was frozen and admitted. Nothing has left the process.
    IntentAdmitted,
    /// Broker authority was obtained and this exact attempt was prepared. This
    /// is the last append before the irreversible boundary.
    DispatchPrepared,
    /// What the evidence says about whether the effect landed.
    OutcomeObserved,
    /// A named predicate was evaluated and produced a result.
    Verified,
}

/// Which journal resource reached its hard admission ceiling.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum JournalLimitKind {
    /// Number of retained events.
    Events,
    /// Encoded frame bytes admitted to the journal history.
    Bytes,
    /// A continuation watermark at or past the ceiling it would be measured
    /// against.
    ContinuationWatermark,
    /// Generations a chain of journals may hold, counted from the name a
    /// continuation is derived from.
    Generation,
    /// Actions one sealed checkpoint may fold.
    CheckpointActions,
    /// Unresolved attempts one sealed checkpoint may carry.
    CheckpointUnresolved,
}

impl EventKind {
    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::IntentAdmitted => "intent_admitted",
            Self::DispatchPrepared => "dispatch_prepared",
            Self::OutcomeObserved => "outcome_observed",
            Self::Verified => "verified",
        }
    }

    /// The kind that may follow this one for the same key.
    ///
    /// One attempt at one intent walks the *effect* ladder exactly once. A retry
    /// is a new [`crate::effect::AttemptId`] and therefore a new key, which is
    /// what stops a second `DispatchPrepared` for an existing key from being
    /// representable at all.
    ///
    /// Verification is the one rung that may repeat. It is a statement about the
    /// world *at the time a predicate ran*, and the world moves: the file is
    /// edited, the remote state changes, the predicate itself is revised. A
    /// latched verdict would stay `Verified` forever with nothing able to say it
    /// no longer holds, so a later `Verified` event supersedes the earlier one and
    /// the recovered status follows the latest verdict. The effect has still
    /// happened once; only what is known about its postcondition is revised.
    #[must_use]
    pub const fn next(self) -> Option<Self> {
        match self {
            Self::IntentAdmitted => Some(Self::DispatchPrepared),
            Self::DispatchPrepared => Some(Self::OutcomeObserved),
            Self::OutcomeObserved | Self::Verified => Some(Self::Verified),
        }
    }
}

impl fmt::Display for EventKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl VerificationResult {
    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Satisfied => "satisfied",
            Self::NotSatisfied => "not_satisfied",
        }
    }
}

impl fmt::Display for VerificationResult {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// A named predicate's answer, with what it was evaluated against.
///
/// The observations are carried as a digest rather than an inline record so the
/// journal stays bounded, and as a digest rather than a summary so the answer
/// cannot be re-read under a different input set. A bare "done" has no place
/// here: it names no predicate, so nothing can re-evaluate it.
#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    Hash,
    lgwks_std::wire::Archive,
    lgwks_std::wire::Serialize,
    lgwks_std::wire::Deserialize,
)]
#[rkyv(crate = lgwks_std::wire::rkyv, compare(PartialEq), derive(Debug))]
pub struct Verification {
    /// Which predicate was evaluated.
    predicate: Id128,
    /// Which revision of that predicate. A changed predicate is a different
    /// claim over the same observations, so the version is part of the record.
    predicate_version: u64,
    /// Digest of the observations the predicate was evaluated against.
    observations: Digest,
    /// The answer.
    result: VerificationResult,
}

impl Verification {
    /// Build a verification record.
    #[must_use]
    pub const fn new(
        predicate: Id128,
        predicate_version: u64,
        observations: Digest,
        result: VerificationResult,
    ) -> Self {
        Self {
            predicate,
            predicate_version,
            observations,
            result,
        }
    }

    /// Which predicate was evaluated.
    #[must_use]
    pub const fn predicate(self) -> Id128 {
        self.predicate
    }

    /// Which revision of that predicate.
    #[must_use]
    pub const fn predicate_version(self) -> u64 {
        self.predicate_version
    }

    /// Digest of the observations the predicate was evaluated against.
    #[must_use]
    pub const fn observations(self) -> Digest {
        self.observations
    }

    /// The answer.
    #[must_use]
    pub const fn result(self) -> VerificationResult {
        self.result
    }
}

impl EffectEvent {
    /// The attempt this fact is about.
    #[must_use]
    pub const fn key(self) -> EffectKey {
        match self {
            Self::IntentAdmitted { key }
            | Self::DispatchPrepared { key }
            | Self::OutcomeObserved { key, .. }
            | Self::Verified { key, .. } => key,
        }
    }

    /// Which kind of fact this is.
    #[must_use]
    pub const fn kind(self) -> EventKind {
        match self {
            Self::IntentAdmitted { .. } => EventKind::IntentAdmitted,
            Self::DispatchPrepared { .. } => EventKind::DispatchPrepared,
            Self::OutcomeObserved { .. } => EventKind::OutcomeObserved,
            Self::Verified { .. } => EventKind::Verified,
        }
    }

    /// The event's encoding, which is what the chain hashes.
    ///
    /// Public because an adapter outside this crate has to compute the same
    /// head to be a journal at all. The encoding is the estate's, not this
    /// module's: the record is archived by [`lgwks_std::wire`] exactly as every
    /// other estate type that crosses a byte boundary, so the discriminant, the
    /// identity and the payload are one encode rather than a framing this module
    /// maintains beside it. A reader accesses the archive in place instead of
    /// decoding it, which is the reason the format exists.
    ///
    /// The bytes are what a chain head commits to, so they are part of this
    /// module's durable contract: a change to the record's shape changes every
    /// head computed after it.
    ///
    /// # Errors
    ///
    /// [`WireError`] if the archive cannot be allocated.
    pub fn to_bytes(self) -> Result<AlignedVec, WireError> {
        lgwks_std::wire::to_bytes::<WireError>(&self)
    }
}

impl fmt::Display for EffectEvent {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{} {}", self.kind(), self.key())
    }
}

/// One committed entry: the event and the position it landed at.
///
/// A named struct rather than a tuple because the two fields answer different
/// questions and a reader should not have to remember which came first.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct JournalEntry {
    /// Where the entry sits in the chain.
    position: JournalPosition,
    /// The fact that was appended.
    event: EffectEvent,
}

impl JournalEntry {
    /// Build an entry.
    #[must_use]
    pub const fn new(position: JournalPosition, event: EffectEvent) -> Self {
        Self { position, event }
    }

    /// Where the entry sits in the chain.
    #[must_use]
    pub const fn position(&self) -> JournalPosition {
        self.position
    }

    /// The fact that was appended.
    ///
    /// By reference: [`EffectEvent`] carries a whole [`EffectKey`], and a
    /// journal replay walks every entry, so copying one per step would be a
    /// 192-byte move for no gain.
    #[must_use]
    pub const fn event(&self) -> &EffectEvent {
        &self.event
    }
}

/// What a journal can refuse.
#[derive(Debug)]
#[non_exhaustive]
pub enum JournalError {
    /// The tail the caller expected is not the tail that is committed.
    ///
    /// Not merged and not retried in place: it means another writer appended, so
    /// the caller's view of the run is stale and it has to re-read before it can
    /// decide anything.
    TailMismatch {
        /// The tail the caller stated.
        expected: JournalPosition,
        /// The tail the journal holds.
        actual: JournalPosition,
    },
    /// Recovery folded one history, but the journal advanced before that
    /// history could become the controller's append fence.
    SnapshotStale {
        /// Number of events recovery folded.
        recovered_events: u64,
        /// Number of events the journal reported afterwards.
        committed_events: u64,
    },
    /// The journal cannot promise the durability the caller requires.
    PromiseUnmet {
        /// What the caller needed.
        required: DurabilityPromise,
        /// What the journal offers.
        offered: DurabilityPromise,
    },
    /// The adapter cannot attest that an already-recorded outcome now meets
    /// the requested durability grade.
    ReceiptUnavailable {
        /// What the caller needs before it can settle the attempt.
        required: DurabilityPromise,
    },
    /// A receipt names a different append than the outcome being settled.
    ReceiptMismatch {
        /// The committed outcome position the caller is settling.
        expected: JournalPosition,
        /// The position the adapter returned in its receipt.
        actual: JournalPosition,
    },
    /// The position exists, but it contains a different event than the one
    /// the caller is trying to acknowledge.
    EntryMismatch {
        /// The exact committed position inspected.
        position: JournalPosition,
        /// The event the caller expected there.
        expected: Box<EffectEvent>,
        /// The event the adapter returned at that position.
        actual: Box<EffectEvent>,
    },
    /// The event cannot follow what is committed for that key.
    OutOfOrder {
        /// The attempt the refused event was about.
        ///
        /// Boxed because [`EffectKey`] is 128 bytes and an error carrying one by
        /// value would make every `Result` in this module pay for a case that is
        /// refused before it changes anything.
        key: Box<EffectKey>,
        /// What the ladder allows next, or `None` when the key is complete.
        expected: Option<EventKind>,
        /// What the caller tried to append.
        attempted: EventKind,
    },
    /// The run has more events than a journal position can address.
    ///
    /// Reachable only after 2^64 appends to one run. Named rather than wrapped,
    /// because a wrapped sequence would make two positions compare equal and
    /// defeat the guard the position exists to provide.
    Exhausted,
    /// A sealed journal's successor is complete, so this journal is not the
    /// authoritative one.
    ///
    /// The refusal carries where the authoritative journal is rather than a
    /// rendered string, because the caller's whole next move is to open that path
    /// and a caller that had to parse a message to learn it would be one message
    /// away from opening the sealed predecessor and appending to it.
    ///
    /// This is never a fault and never a loss: the predecessor's committed history
    /// is intact and readable, and the successor carries every attempt it left
    /// unresolved.
    Superseded {
        /// Where the authoritative journal lives.
        path: String,
    },
    /// An append names an attempt of an action the journal's chain already walked.
    ///
    /// Distinct from [`Self::OutOfOrder`] because it is a different fact. An
    /// out-of-order append names a rung that cannot follow a rung the journal
    /// holds *for that exact key*. This names an attempt number at or below the
    /// one a continuation folded for its action, which is a replay of an attempt
    /// the journal sealed: the key itself may never have been carried, because
    /// the sealed history, not the checkpoint, is what remembers it. Collapsing
    /// the two would report a replay of a sealed attempt as a ladder mistake on a
    /// key the journal has never heard of.
    AttemptAlreadyWalked {
        /// The attempt the refused event was about.
        key: Box<EffectKey>,
        /// The latest attempt the journal's chain walked for that action.
        latest: crate::effect::AttemptId,
    },
    /// A continuation was armed to stop at a boundary, and did.
    ///
    /// The fault-injection door a crash harness kills a process on. Nothing was
    /// acknowledged and the journal will not append again, so a caller that sees
    /// this reopens and reads the file back; a caller that was not expecting it
    /// sees a continuation that did not happen rather than a journal that quietly
    /// continued.
    ContinuationPaused {
        /// The boundary the continuation stopped at.
        boundary: SealPause,
    },
    /// Appending or opening would exceed a shipped journal's hard history cap.
    ///
    /// The refusal is non-destructive: committed and unresolved evidence is
    /// retained, and an over-limit file is not truncated to make it fit.
    CapacityExceeded {
        /// Which finite resource would exceed its ceiling.
        resource: JournalLimitKind,
        /// The resource's configured hard ceiling.
        limit: u64,
        /// The amount required by the refused operation or existing file.
        requested: u64,
    },
    /// Another writer holds the journal file's exclusive fence.
    ///
    /// The refusal carries the path because the fix is about *when* to open,
    /// not what to fix: wait for the holder to finish or die, then reopen.
    /// Appending under a second unfenced handle would let two controllers
    /// acknowledge histories that cannot replay as one chain, so this is
    /// refused before a single byte of the file is read — the replay and the
    /// torn-tail repair are the owner's alone.
    ///
    /// The fence is the operating system's advisory file lock: it binds
    /// cooperating writers that go through this constructor, and it is
    /// released by the kernel when the holder closes the file or dies. It is
    /// not containment against a writer that never asks for the lock, and it
    /// does not coordinate writers on other hosts — a distributed lease is a
    /// different mechanism.
    Locked {
        /// Where the contended journal lives.
        path: String,
    },
    /// The backing store refused the append before the event could be written.
    ///
    /// The journal is unchanged: the event is not committed, and a retry may
    /// re-append. Named apart from [`Self::OutcomeUnknown`] so a caller can
    /// tell "nothing landed, try again" from "something may have landed,
    /// read back" without knowing the adapter.
    Storage(io::Error),
    /// The append may have committed, and the acknowledgment was lost.
    ///
    /// This is the reply a real store produces when bytes reach the disk and
    /// the confirmation does not reach the caller: a write that fails part
    /// way, a sync whose result never arrives. The committed state of the
    /// event is *unknown* — it may be in the store, it may not — and the
    /// contract for a caller is reconciliation by readback, never a blind
    /// re-append: re-appending an event that did land is how an effect gets
    /// executed twice on the strength of a lost reply.
    ///
    /// An adapter returns this only for its own append's uncertain outcome.
    /// Every other refusal — a stale tail, a ladder violation, a store that
    /// refused before touching the log — leaves the journal unchanged and is
    /// reported as its own variant.
    OutcomeUnknown {
        /// The store error the reply was lost behind, when there was one.
        cause: io::Error,
    },
    /// The event could not be encoded for the chain.
    ///
    /// Not a caller error and not reachable by anything the caller controls:
    /// the encoding allocates, and this is that allocation failing. Named
    /// rather than folded into [`Self::Storage`] because nothing was stored —
    /// the append is refused before the journal is touched at all.
    Encoding(WireError),
    /// The journal's committed bytes do not re-derive the history they claim.
    ///
    /// This is not an interrupted append — an interrupted append is a torn
    /// tail, and a torn tail was never acknowledged, so repairing it costs
    /// nothing. This is committed bytes that no longer mean what the chain
    /// says: bit rot, or a hand on the file. It is refused rather than
    /// trimmed, because the record may have been acknowledged, and an
    /// acknowledgment the journal quietly rewrites is not a record.
    ///
    /// Boxed for the same reason [`Self::OutOfOrder`] boxes its key: an error
    /// carrying the payload by value would make every `Result` in this module
    /// pay for a case that is refused before it changes anything.
    Corrupt(Box<Corruption>),
}

impl fmt::Display for JournalError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // The scrutinee is `*self` so each pattern's type is the enum's own type
        // rather than a reference to it, and the two non-`Copy` payloads are
        // bound by `ref`. `clippy::pattern_type_mismatch` is forbidden in this
        // workspace, and this is the form it asks for.
        match *self {
            Self::TailMismatch { expected, actual } => write!(
                f,
                "journal tail moved: expected {expected}, committed {actual}"
            ),
            Self::SnapshotStale {
                recovered_events,
                committed_events,
            } => write!(
                f,
                "journal advanced during recovery: folded {recovered_events} events, committed {committed_events}"
            ),
            Self::PromiseUnmet { required, offered } => write!(
                f,
                "journal promises {offered}, which is below the {required} this needs"
            ),
            Self::ReceiptUnavailable { required } => write!(
                f,
                "journal cannot attest that the recorded outcome meets {required}"
            ),
            Self::ReceiptMismatch { expected, actual } => write!(
                f,
                "journal receipt names {actual}, not the outcome committed at {expected}"
            ),
            Self::EntryMismatch {
                position,
                ref expected,
                ref actual,
            } => write!(
                f,
                "journal entry at {position} is {actual}, not the acknowledged {expected}"
            ),
            Self::OutOfOrder {
                ref key,
                expected,
                attempted,
            } => match expected {
                Some(next) => write!(
                    f,
                    "{attempted} cannot follow what is committed for {key}; \
                     the next recorded fact is {next}"
                ),
                None => write!(
                    f,
                    "{attempted} cannot be appended for {key}; \
                     its recorded facts are complete"
                ),
            },
            Self::Exhausted => f.write_str("journal position exhausted"),
            Self::Superseded { ref path } => write!(
                f,
                "this journal is sealed and its successor is authoritative; open {path}"
            ),
            Self::AttemptAlreadyWalked { ref key, latest } => write!(
                f,
                "{key} is an attempt at or below the latest this journal walked \
                 for its action, which is {latest}; the sealed history already \
                 records it"
            ),
            Self::ContinuationPaused { boundary } => write!(
                f,
                "the continuation was armed to stop at {boundary}, so no successor \
                 was established"
            ),
            Self::CapacityExceeded {
                resource,
                limit,
                requested,
            } => write!(
                f,
                "journal {resource:?} limit exceeded: limit {limit}, requested {requested}"
            ),
            Self::Locked { ref path } => write!(
                f,
                "journal at {path} is held by another writer; \
                 wait for the holder to finish, then reopen"
            ),
            Self::Storage(ref cause) => {
                write!(f, "journal storage refused the append: {cause}")
            }
            Self::OutcomeUnknown { ref cause } => write!(
                f,
                "journal append may have committed, acknowledgment lost: {cause}; \
                 reconcile by readback rather than re-appending"
            ),
            Self::Encoding(ref cause) => {
                write!(
                    f,
                    "journal could not encode the event for the chain: {cause}"
                )
            }
            Self::Corrupt(ref corruption) => {
                write!(f, "journal refused its own committed bytes: {corruption}")
            }
        }
    }
}

impl std::error::Error for JournalError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match *self {
            Self::Storage(ref cause) => Some(cause),
            Self::OutcomeUnknown { ref cause } => Some(cause),
            Self::Encoding(ref cause) => Some(cause),
            Self::Corrupt(ref cause) => Some(cause),
            _ => None,
        }
    }
}

/// Proof that an append is committed under a named durability promise.
///
/// An acknowledgment is not a buffer acceptance, a log line or a `Drop`. It is
/// the statement that the configured promise has been met and that the journal's
/// head is now at [`Self::position`]. An adapter mints this only after the
/// promise is met, and nothing here can check that it did, which is why the
/// promise travels on the acknowledgment: a caller can at least see what was
/// claimed, and a test against a real store is what decides whether the claim
/// was true.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct DurableAck {
    /// The committed position, sequence and head together.
    position: JournalPosition,
    /// The promise the journal claims to have met.
    promise: DurabilityPromise,
}

impl DurableAck {
    /// Mint an acknowledgment.
    ///
    /// Public because an adapter implementing [`EffectJournal`] outside this
    /// crate is where an acknowledgment is produced, and the type cannot be
    /// built any other way.
    #[must_use]
    pub const fn new(position: JournalPosition, promise: DurabilityPromise) -> Self {
        Self { position, promise }
    }

    /// Where the append landed.
    #[must_use]
    pub const fn position(self) -> JournalPosition {
        self.position
    }

    /// What the journal claims about surviving a crash.
    #[must_use]
    pub const fn promise(self) -> DurabilityPromise {
        self.promise
    }
}

/// An append-only record of a run's effects.
///
/// Implementors are adapters over a host's durable store. The trait is the
/// seam; the durability grade is what a caller checks before it uses one.
pub trait EffectJournal {
    /// What this journal promises about an acknowledged append.
    fn durability(&self) -> DurabilityPromise;

    /// The last committed position, or the genesis when nothing is committed.
    fn tail(&self) -> JournalPosition;

    /// Every committed event, in commit order.
    ///
    /// Required rather than optional, and read rather than streamed, because
    /// this is what [`recover`] consumes: a controller that comes back after a
    /// crash learns what was in flight by replaying its own journal, and a
    /// journal that cannot be read back cannot be the record behind an external
    /// handoff. An adapter over a store that can only be appended to is a
    /// *sink*, not a journal, and this trait is not a sink.
    ///
    /// The whole sequence rather than a page, because [`recover`] folds over it
    /// and a partial replay answers "what is uncertain" with a subset — which
    /// reads as "nothing is uncertain" for every attempt in the part that was
    /// not read.
    ///
    /// Shipped adapters enforce [`MAX_JOURNAL_EVENTS`] and
    /// [`MAX_JOURNAL_BYTES`]. They refuse new history at the cap instead of
    /// deleting old or unresolved events. External adapters remain responsible
    /// for bounding their own backing store and returned vector.
    ///
    /// # Errors
    ///
    /// [`JournalError::Storage`] when the backing store refused to be read.
    fn committed(&self) -> Result<Vec<EffectEvent>, JournalError>;

    /// Every committed entry, including the position each event occupies.
    ///
    /// An acknowledgement is meaningful only when positioned readback proves
    /// it names the exact appended fact. A later tail can belong to another
    /// controller and must never become this controller's append fence.
    ///
    /// Adapters that cannot provide positioned readback must refuse dispatch,
    /// including local effects, rather than authorize an unbound append.
    fn committed_entries(&self) -> Result<Vec<JournalEntry>, JournalError> {
        Err(JournalError::ReceiptUnavailable {
            required: self.durability(),
        })
    }

    /// Read one position-bound entry without returning the whole committed
    /// history.
    ///
    /// The default preserves correctness for external adapters while providing
    /// a migration path; shipped journals override it with sequence-indexed
    /// access. Callers still compare the returned event with the exact event
    /// they acknowledged, because position alone is not event identity.
    fn committed_entry(
        &self,
        position: JournalPosition,
    ) -> Result<Option<JournalEntry>, JournalError> {
        Ok(self
            .committed_entries()?
            .into_iter()
            .find(|entry| entry.position() == position))
    }

    /// Find the latest committed outcome for one attempt, with its position.
    ///
    /// This default is a compatibility path for external adapters. Shipped
    /// journals maintain a keyed index so repeated reconciliation does not
    /// clone and scan all historical events.
    fn outcome_at(
        &self,
        key: EffectKey,
    ) -> Result<Option<(JournalPosition, EffectEvidence)>, JournalError> {
        Ok(self
            .committed_entries()?
            .into_iter()
            .rev()
            .find_map(|entry| match *entry.event() {
                EffectEvent::OutcomeObserved {
                    key: held,
                    evidence,
                } if held == key => Some((entry.position(), evidence)),
                _ => None,
            }))
    }

    /// Append `event` if and only if `expected_tail` is still the committed
    /// tail.
    ///
    /// The `&mut self` receiver fences writers within one process; the tail
    /// check fences writers across processes, which is the case that actually
    /// happens.
    ///
    /// The outcome is one of exactly three things, and an adapter must say
    /// which:
    ///
    /// - `Ok(ack)` — the event is committed at `ack.position` under
    ///   `ack.promise`.
    /// - `Err` of [`JournalError::TailMismatch`], [`JournalError::OutOfOrder`],
    ///   [`JournalError::Exhausted`], [`JournalError::Encoding`] or
    ///   [`JournalError::Storage`] — the event is *not* committed and the
    ///   journal is unchanged. A retry may re-append.
    /// - `Err(JournalError::OutcomeUnknown)` — the event *may* be committed
    ///   and the acknowledgment was lost. A caller reconciles by readback and
    ///   must not re-append: re-appending an event that did land is how an
    ///   effect gets executed twice on the strength of a lost reply.
    ///
    /// # Errors
    ///
    /// [`JournalError::TailMismatch`] when another writer appended,
    /// [`JournalError::OutOfOrder`] when the event cannot follow what is
    /// committed for its key, [`JournalError::Exhausted`] when the position
    /// space is spent, [`JournalError::Storage`] when the backing store
    /// refused before the event was written, and
    /// [`JournalError::OutcomeUnknown`] when the event may have committed and
    /// the acknowledgment was lost.
    fn compare_and_append(
        &mut self,
        expected_tail: JournalPosition,
        event: &EffectEvent,
    ) -> Result<DurableAck, JournalError>;

    /// The same append, for a caller that awaits rather than blocks.
    ///
    /// One contract with two doors, and the reason is the storage device's
    /// cost. A store whose durability path blocks — a `fsync` on the caller's
    /// own thread — charges that latency to everything else sharing the
    /// executor, and awaiting the surrounding tick does not make the call
    /// inside it non-blocking. An adapter that owns its storage on a separate
    /// thread overrides this and returns the same acknowledgment, from the
    /// same fence, over the same bytes.
    ///
    /// The default reaches the synchronous form, which is correct for a store
    /// with no blocking durability path. It is a default rather than a
    /// requirement so an adapter is not asked to reimplement the fence; what
    /// it must not do is answer with a weaker write than its synchronous form
    /// would have made.
    ///
    /// Dropping the returned future is not a cancellation of the append. An
    /// adapter whose append can outlive its caller has to leave the handle in
    /// a state that refuses the next append rather than continue from a view
    /// that may be behind the store.
    ///
    /// # Errors
    ///
    /// Every [`JournalError`] [`Self::compare_and_append`] produces, from the
    /// same checks in the same order.
    fn compare_and_append_async<'a>(
        &'a mut self,
        expected_tail: JournalPosition,
        event: &'a EffectEvent,
    ) -> BoxFuture<'a, Result<DurableAck, JournalError>> {
        Box::pin(async move { self.compare_and_append(expected_tail, event) })
    }

    /// Confirm that an existing outcome record now meets `required`.
    ///
    /// An append can return a weak acknowledgment after it has already moved
    /// the per-key ladder to `OutcomeObserved`. A retry cannot append that
    /// outcome again, so an adapter that can flush, replicate, or otherwise
    /// obtain a stronger receipt implements this method. The default refuses
    /// rather than treating read-back of event bytes as a durability proof.
    ///
    /// # Errors
    ///
    /// [`JournalError::ReceiptUnavailable`] when the adapter has no receipt
    /// operation, or another journal error when it cannot obtain one.
    fn confirm_outcome(
        &mut self,
        _key: EffectKey,
        _evidence: EffectEvidence,
        _position: JournalPosition,
        required: DurabilityPromise,
    ) -> Result<DurableAck, JournalError> {
        Err(JournalError::ReceiptUnavailable { required })
    }

    /// Whether this journal may host an effect that has left the process.
    ///
    /// A journal that cannot survive its own writer dying must never be the
    /// record behind an external handoff, because after a crash it reports an
    /// empty run rather than an uncertain one.
    ///
    /// # Errors
    ///
    /// [`JournalError::PromiseUnmet`] when the journal is ephemeral.
    fn admit_external_handoff(&self) -> Result<DurabilityPromise, JournalError> {
        let offered = self.durability();
        if offered.survives_process_crash() {
            return Ok(offered);
        }
        Err(JournalError::PromiseUnmet {
            required: DurabilityPromise::ProcessCrash,
            offered,
        })
    }

    /// Whether this journal can admit `rungs` more events for an attempt it is
    /// about to hand off, decided *before* any of them is written.
    ///
    /// A handoff records at least three rungs — `IntentAdmitted`,
    /// `DispatchPrepared` and the settling `OutcomeObserved` — and the last one
    /// is written only after the effect has left the process. A journal that
    /// admits the first two and then refuses the third leaves an attempt that
    /// can never be settled without deleting unresolved evidence: the exact
    /// availability loss the hard ceiling must not cause. A bounded journal
    /// answers this by checking the whole attempt fits, so a refusal happens
    /// before the handoff rather than after it.
    ///
    /// The default is permissive, so external adapters keep their current
    /// behaviour; a bounded adapter overrides it. The check is a conservative
    /// admission test, not a held reservation: it is made on the serialized
    /// append path, where the count it reads cannot move between the check and
    /// the writes it authorizes.
    ///
    /// # Errors
    ///
    /// [`JournalError::CapacityExceeded`] when the attempt does not fit.
    fn reserve_handoff_capacity(&self, _rungs: u64) -> Result<(), JournalError> {
        Ok(())
    }

    /// How much of each ceiling this journal has used, and whether it continues
    /// past them at all.
    ///
    /// The question a caller asks before an append, and it is asked on the shipped
    /// append path rather than by a caller a test wrote, because a lifecycle
    /// policy nobody consults is a policy that does not exist (issue #267).
    ///
    /// The default measures the journal honestly and reports `continues() ==
    /// false`, which leaves every adapter's behaviour exactly as it was: the
    /// default is what the in-memory adapter and every external adapter want,
    /// and a `FileJournal` opened for an unattended run is the one that answers
    /// otherwise.
    ///
    /// # Errors
    ///
    /// [`JournalError::Storage`] when the store cannot be read for the counts.
    fn continuation_watermark(&self) -> Result<ContinuationWatermark, JournalError> {
        Ok(ContinuationWatermark::inert(
            0,
            u64::saturating_from(MAX_JOURNAL_EVENTS),
            0,
            MAX_JOURNAL_BYTES,
        ))
    }

    /// Seal this journal at the watermark and hand back the successor opened from
    /// the checkpoint this writes.
    ///
    /// `None` when this journal does not continue, which is the default and the
    /// answer every adapter but the continuing file journal gives. A `Some` hands
    /// the caller the authoritative journal: the predecessor is sealed, read-only
    /// and still readable, and every append from here lands in the successor.
    ///
    /// The caller must adopt the returned journal whole — its tail, its ladder and
    /// its carried state — rather than keeping the predecessor's own position,
    /// because the successor's chain starts at the predecessor's tail and its
    /// first frame is the seal frame. An append that kept the predecessor's fence
    /// would be refused as a stale tail, which is the correct answer to a
    /// question that must not be asked.
    ///
    /// # Errors
    ///
    /// sealed-handle refusals: [`JournalError::Superseded`] when
    /// this journal is already sealed, [`JournalError::CapacityExceeded`] when
    /// the carried state is past a declared bound, and the device's own errors
    /// through [`JournalError::OutcomeUnknown`] when the bytes may be on the disk
    /// under no acknowledgment.
    fn continue_as_new(&mut self) -> Result<Option<Box<dyn EffectJournal>>, JournalError> {
        Ok(None)
    }
}

/// What is known about one attempt after replaying a journal.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum AttemptStatus {
    /// The intent was admitted and nothing was handed to an external system.
    Prepared,
    /// A dispatch was prepared and no outcome was ever recorded. This is the
    /// state a crash between the prepare append and the result recovers into,
    /// and it is deliberately not `NotApplied`: nothing established that the
    /// bytes did not arrive, and assuming they did not is how a duplicate
    /// non-idempotent effect gets sent.
    OutcomeUnknown,
    /// Evidence says the effect landed.
    Applied,
    /// Evidence says the effect did not land.
    NotApplied,
    /// A named predicate was evaluated and held over observations newer than
    /// the effect.
    Verified,
    /// The effect landed, and a named predicate was evaluated and did **not**
    /// hold.
    ///
    /// Distinct from [`Applied`](Self::Applied), which is also what an attempt
    /// reads as before any predicate has run: folding the two together made a
    /// failed verification indistinguishable from an unverified success, and the
    /// failure recoverable only by re-deriving it from the raw events, which is a
    /// second authority. The effect is still known to have happened, so a
    /// recovery path must not resend it.
    VerificationFailed,
}

impl AttemptStatus {
    /// Whether an external system may or may not have acted.
    ///
    /// The one predicate a recovery path should branch on before it considers
    /// any resend.
    #[must_use]
    pub const fn is_uncertain(self) -> bool {
        matches!(self, Self::OutcomeUnknown)
    }

    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Prepared => "prepared",
            Self::OutcomeUnknown => "outcome_unknown",
            Self::Applied => "applied",
            Self::NotApplied => "not_applied",
            Self::Verified => "verified",
            Self::VerificationFailed => "verification_failed",
        }
    }
}

impl fmt::Display for AttemptStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// One attempt and what the journal knows about it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Attempt {
    /// The attempt.
    key: EffectKey,
    /// What is known about it.
    status: AttemptStatus,
    /// The latest verification the journal holds for it, which is what a
    /// `Verified` or `VerificationFailed` status is qualified by: which predicate,
    /// at which version, over which observations.
    verification: Option<Verification>,
}

impl Attempt {
    /// Build an attempt record.
    #[must_use]
    pub const fn new(key: EffectKey, status: AttemptStatus) -> Self {
        Self {
            key,
            status,
            verification: None,
        }
    }

    /// The latest verification recorded for this attempt, or `None` when no
    /// predicate has been evaluated.
    ///
    /// A `Verified` status is only as good as the predicate version and
    /// observation digest it was decided at; this is how a consumer reads them
    /// and compares them with the state it is about to act on.
    #[must_use]
    pub const fn verification(&self) -> Option<Verification> {
        self.verification
    }

    /// The attempt.
    #[must_use]
    pub const fn key(&self) -> EffectKey {
        self.key
    }

    /// What is known about it.
    #[must_use]
    pub const fn status(&self) -> AttemptStatus {
        self.status
    }
}

/// One change of an attempt's recovered status.
///
/// `position` is the event's place in the sequence that was folded, counted from
/// zero. It is a *journal position*, not a wall-clock time, and that is
/// deliberate: recovery is a pure fold that two replays must agree on, and a
/// clock read inside it would make them disagree. "When" is answered by the
/// order of facts, which is the only order a replayed journal has.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Transition {
    /// The status before this change, or `None` for the attempt's first event.
    from: Option<AttemptStatus>,
    /// The status after it.
    to: AttemptStatus,
    /// The event's position in the folded sequence.
    position: usize,
}

impl Transition {
    /// The status this change superseded, or `None` for the first.
    #[must_use]
    pub const fn from(&self) -> Option<AttemptStatus> {
        self.from
    }

    /// The status this change produced.
    #[must_use]
    pub const fn to(&self) -> AttemptStatus {
        self.to
    }

    /// Where in the folded sequence the change happened.
    #[must_use]
    pub const fn position(&self) -> usize {
        self.position
    }
}

/// Every attempt a journal's events mention, in the order their intent was
/// first admitted.
///
/// Ordered by first appearance rather than by key, because key order is
/// arbitrary and a report a person reads should list attempts in the order the
/// run reached them.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Recovered {
    /// One entry per key, in admission order.
    attempts: Vec<Attempt>,
    /// Direct lookup by key while preserving `attempts` as the report order.
    index: HashMap<EffectKey, usize>,
    /// Every status change of each attempt, parallel to `attempts`. Bounded by
    /// the number of events folded, which the journal's own ceiling bounds.
    transitions: Vec<Vec<Transition>>,
}

impl Recovered {
    /// What is known about `key`, or `None` when the journal never mentioned
    /// it.
    #[must_use]
    pub fn status(&self, key: EffectKey) -> Option<AttemptStatus> {
        self.index
            .get(&key)
            .and_then(|index| self.attempts.get(*index))
            .map(|attempt| attempt.status)
    }

    /// The latest verification recorded for `key`, or `None` when the journal
    /// never mentioned it or no predicate has run.
    #[must_use]
    pub fn verification(&self, key: EffectKey) -> Option<Verification> {
        self.index
            .get(&key)
            .and_then(|index| self.attempts.get(*index))
            .and_then(Attempt::verification)
    }

    /// Every change of `key`'s status in the order it happened, empty when the
    /// journal never mentioned it.
    ///
    /// The answer to "what changed", beside [`status`](Self::status)'s "what is
    /// true now": the current status stays a cheap closed fold, and the history
    /// is separate rather than folded into it.
    #[must_use]
    pub fn history(&self, key: EffectKey) -> &[Transition] {
        self.index
            .get(&key)
            .and_then(|index| self.transitions.get(*index))
            .map_or(&[], Vec::as_slice)
    }

    /// How many attempts the journal mentions.
    #[must_use]
    pub fn len(&self) -> usize {
        self.attempts.len()
    }

    /// Whether the journal mentions no attempts.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.attempts.is_empty()
    }

    /// The attempts whose external outcome is unknown.
    ///
    /// This is the list a recovery path must settle before it may dispatch
    /// anything again, so it is a first-class query rather than a fold each
    /// caller writes for itself.
    #[must_use]
    pub fn uncertain(&self) -> Vec<EffectKey> {
        self.attempts
            .iter()
            .filter(|attempt| attempt.status.is_uncertain())
            .map(|attempt| attempt.key)
            .collect()
    }

    /// Every attempt and its state, in admission order.
    #[must_use]
    pub fn attempts(&self) -> &[Attempt] {
        &self.attempts
    }
}

/// Fold a journal's events into what is known about each attempt.
///
/// The rule that makes this worth having: a `DispatchPrepared` with no later
/// outcome recovers as [`AttemptStatus::OutcomeUnknown`], never as
/// `NotApplied`. A controller that cannot positively prove no handoff occurred
/// has to stay uncertain, and losing availability is the cheaper error.
#[must_use]
pub fn recover<'a>(events: impl IntoIterator<Item = &'a EffectEvent>) -> Recovered {
    recover_continued(None, events)
}

/// Fold a journal that may begin from a sealed checkpoint.
///
/// [`recover`] with the carried state of a successor folded in first, which is
/// what makes the answer *the same* on both sides of a continuation: the
/// predecessor reports an attempt from its own suffix, the successor reports it
/// from the checkpoint that carried it, and a caller cannot tell which journal
/// answered except by looking at the path.
///
/// Two things are seeded, and the difference matters. An **unresolved** attempt is
/// seeded as its own key at its own rung, so it reads back exactly as uncertain as
/// it was and the ladder refuses a resend. A **settled** attempt is seeded as the
/// action's latest key, so the status and the verification a caller compares
/// against survive the boundary; the older attempts that key stands for are not
/// carried, and are refused instead — `AttemptId` is monotonic per action, so the
/// folded record answers "was this walked?" without the checkpoint holding every
/// attempt ever.
///
/// A seed whose rung or status this build does not name is **skipped**, and the
/// successor's own open refuses the checkpoint rather than reading it. Reading an
/// unknown rung as the foot of the ladder would make a sealed attempt replayable,
/// and reading an unknown status as absent would make a verified attempt invisible.
#[must_use]
pub fn recover_continued<'a>(
    base: Option<&Continuation>,
    events: impl IntoIterator<Item = &'a EffectEvent>,
) -> Recovered {
    let mut recovered = Recovered::default();
    if let Some(checkpoint) = base {
        for carried in checkpoint.settled() {
            let (Some(status), Some(_rung)) = (carried.status(), carried.rung()) else {
                continue;
            };
            seed(
                &mut recovered,
                carried.key(),
                status,
                carried.verification(),
            );
        }
        for carried in checkpoint.unresolved() {
            let Some(status) = carried.recovered_status() else {
                continue;
            };
            seed(&mut recovered, carried.key(), status, None);
        }
    }
    fold_events(&mut recovered, events);
    recovered
}

/// Put one carried attempt into a fold, in the order the checkpoint lists them.
fn seed(
    recovered: &mut Recovered,
    key: EffectKey,
    status: AttemptStatus,
    verification: Option<Verification>,
) {
    let index = recovered.attempts.len();
    let mut attempt = Attempt::new(key, status);
    attempt.verification = verification;
    recovered.attempts.push(attempt);
    recovered.transitions.push(vec![Transition {
        from: None,
        to: status,
        position: index,
    }]);
    recovered.index.insert(key, index);
}

/// Fold events over a recovery, one at a time.
fn fold_events<'a>(recovered: &mut Recovered, events: impl IntoIterator<Item = &'a EffectEvent>) {
    for (position, event) in events.into_iter().enumerate() {
        let status = match *event {
            EffectEvent::IntentAdmitted { .. } => AttemptStatus::Prepared,
            EffectEvent::DispatchPrepared { .. } => AttemptStatus::OutcomeUnknown,
            EffectEvent::OutcomeObserved { evidence, .. } => match evidence {
                EffectEvidence::Applied => AttemptStatus::Applied,
                EffectEvidence::NotApplied => AttemptStatus::NotApplied,
            },
            EffectEvent::Verified { verification, .. } => match verification.result() {
                VerificationResult::Satisfied => AttemptStatus::Verified,
                VerificationResult::NotSatisfied => AttemptStatus::VerificationFailed,
            },
        };
        let verification = match *event {
            EffectEvent::Verified { verification, .. } => Some(verification),
            _ => None,
        };
        let key = event.key();
        match recovered.index.get(&key).copied() {
            Some(index) => {
                if let Some(attempt) = recovered.attempts.get_mut(index) {
                    if let Some(trail) = recovered.transitions.get_mut(index) {
                        trail.push(Transition {
                            from: Some(attempt.status),
                            to: status,
                            position,
                        });
                    }
                    attempt.status = status;
                    if verification.is_some() {
                        attempt.verification = verification;
                    }
                }
            }
            None => {
                let index = recovered.attempts.len();
                let mut attempt = Attempt::new(key, status);
                attempt.verification = verification;
                recovered.attempts.push(attempt);
                recovered.transitions.push(vec![Transition {
                    from: None,
                    to: status,
                    position,
                }]);
                recovered.index.insert(key, index);
            }
        }
    }
}

/// Where a recomputed chain stopped agreeing with what was recorded, or why it
/// could not be recomputed at all.
///
/// Two arms rather than one, because encoding the events is a fallible step
/// standing between them and the comparison, and a chain that could not be
/// walked is *unknown* rather than sound. Folding that into "no disagreement
/// found" is the one answer this type must not give.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ChainBreak {
    /// The recomputed head did not match what was recorded.
    Disagreement {
        /// The sequence number of the first entry that did not follow.
        at: u64,
        /// The position the journal recorded for that entry.
        recorded: JournalPosition,
        /// The position recomputed from the events.
        recomputed: JournalPosition,
    },
    /// The entry could not be re-encoded, so the chain could not be checked.
    ///
    /// Nothing was compared. The encoding allocates, and this is that
    /// allocation failing.
    Unencodable {
        /// The sequence number of the entry that could not be encoded.
        at: u64,
    },
}

impl ChainBreak {
    /// The sequence number of the entry the break is about.
    #[must_use]
    pub const fn at(self) -> u64 {
        match self {
            Self::Disagreement { at, .. } | Self::Unencodable { at } => at,
        }
    }

    /// The position the journal recorded, when the break is a disagreement.
    #[must_use]
    pub const fn recorded(self) -> Option<JournalPosition> {
        match self {
            Self::Disagreement { recorded, .. } => Some(recorded),
            Self::Unencodable { .. } => None,
        }
    }

    /// The position recomputed from the events, when the break is a
    /// disagreement.
    #[must_use]
    pub const fn recomputed(self) -> Option<JournalPosition> {
        match self {
            Self::Disagreement { recomputed, .. } => Some(recomputed),
            Self::Unencodable { .. } => None,
        }
    }
}

impl fmt::Display for ChainBreak {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match *self {
            Self::Disagreement {
                at,
                recorded,
                recomputed,
            } => write!(
                f,
                "journal chain breaks at {at}: recorded {recorded}, recomputed {recomputed}"
            ),
            Self::Unencodable { at } => write!(
                f,
                "journal entry {at} could not be re-encoded, so the chain could not be checked"
            ),
        }
    }
}

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

/// The head that follows `previous` once `event` is appended.
///
/// The previous head is hashed in, so the chain commits to order and not only
/// to content: the same two events in the other order produce a different head,
/// which is what makes a reordered or dropped entry detectable rather than
/// merely suspicious.
fn chain(previous: JournalPosition, event: &EffectEvent) -> Result<Digest, JournalError> {
    let mut hasher = Hasher::new();
    hasher.update(previous.head().as_bytes());
    hasher.update(&event.to_bytes().map_err(JournalError::Encoding)?);
    Ok(hasher.finalize())
}

/// [`chain`] over bytes the caller has already archived.
///
/// The two-step split exists because the shared frame step archives an event once
/// and then needs its head: calling [`chain`] there would encode the same value
/// twice and hand the head a second chance to disagree with the bytes that were
/// actually framed. Infallible, because the bytes are already in hand and nothing
/// can fail.
pub(crate) fn chain_over_bytes(previous: &Digest, payload: &[u8]) -> Digest {
    let mut hasher = Hasher::new();
    hasher.update(previous.as_bytes());
    hasher.update(payload);
    hasher.finalize()
}

/// What the ladder allows next for `key`, given the entries committed so far.
///
/// Both shipped adapters keep a per-key index beside the sequence, so this
/// walk is what an index *means* rather than what an append *does*: the last
/// kind recorded for the key names the next allowed one, and a key never
/// seen is at the ladder's foot.
fn next_allowed_of(last: Option<EventKind>) -> Option<EventKind> {
    last.map_or(Some(EventKind::IntentAdmitted), EventKind::next)
}

/// Refuse an append whose tail is stale or whose kind does not follow what is
/// already committed for its key.
///
/// The two checks every adapter's append makes before it touches storage, once:
/// the memory journal and the file journal used to spell them out separately,
/// which is how two stores could come to disagree about what an out-of-order
/// append is. `last` is the last kind committed for the event's key.
///
/// # Errors
///
/// `JournalError::TailMismatch` when `expected_tail` is not `actual`, and
/// `JournalError::OutOfOrder` when the event's kind is not the next one the
/// ladder allows for its key.
fn check_append_order(
    expected_tail: JournalPosition,
    actual: JournalPosition,
    event: &EffectEvent,
    last: Option<EventKind>,
) -> Result<(), JournalError> {
    if expected_tail != actual {
        let refusal = Err(JournalError::TailMismatch {
            expected: expected_tail,
            actual,
        });
        lgwks_std::trace::debug!(error = ?refusal.as_ref().err(), "check_append_order: the caller's view of the tail is stale");
        return refusal;
    }
    let attempted = event.kind();
    let expected = next_allowed_of(last);
    if expected != Some(attempted) {
        let refusal = Err(JournalError::OutOfOrder {
            key: Box::new(event.key()),
            expected,
            attempted,
        });
        lgwks_std::trace::debug!(error = ?refusal.as_ref().err(), "check_append_order: the event does not follow the ladder");
        return refusal;
    }
    Ok(())
}

/// Recompute the chain over `entries` and report the first disagreement.
///
/// # Errors
///
/// [`ChainBreak`] naming the first entry whose recorded position is not the one
/// its events produce.
pub fn verify_chain(entries: &[JournalEntry]) -> Result<JournalPosition, ChainBreak> {
    verify_chain_from(JournalPosition::genesis(), entries)
}

/// [`verify_chain`] over a chain that starts at `base` rather than at the genesis.
///
/// A successor's history begins at its predecessor's chain head, not at the
/// genesis: the seal frame it was opened from is not an event and is not in
/// `committed_entries`, so folding from the genesis would report a disagreement at
/// the very first entry. `base` is what the handle's
/// [`FileJournal::base`](crate::journal::FileJournal::base) reports, which is the
/// position its own scan already verified every frame against.
///
/// # Errors
///
/// [`ChainBreak`] naming the first entry whose recorded position is not the one
/// its events produce.
pub fn verify_chain_from(
    base: JournalPosition,
    entries: &[JournalEntry],
) -> Result<JournalPosition, ChainBreak> {
    let mut position = base;
    for entry in entries {
        let sequence = position.sequence().saturating_add(1);
        let head = match chain(position, entry.event()) {
            Ok(head) => head,
            Err(_) => {
                let refusal = Err(ChainBreak::Unencodable { at: sequence });
                lgwks_std::trace::debug!(error = ?refusal.as_ref().err(), "verify_chain: returning an error to the caller");
                return refusal;
            }
        };
        let recomputed = JournalPosition { sequence, head };
        // The recorded head may have been written by an adversary, so the
        // comparison is the named constant-time one rather than an ordering.
        let recorded = entry.position();
        if recorded.sequence() != sequence || !recorded.head().ct_eq(&head) {
            let refusal = Err(ChainBreak::Disagreement {
                at: sequence,
                recorded,
                recomputed,
            });
            lgwks_std::trace::debug!(error = ?refusal.as_ref().err(), "verify_chain: returning an error to the caller");
            return refusal;
        }
        position = recomputed;
    }
    Ok(position)
}

/// An in-memory journal, for tests and for runs whose effects never leave the
/// process.
///
/// It retains at most [`MAX_JOURNAL_EVENTS`] events and refuses the next append
/// without changing committed state. It reports [`DurabilityPromise::Ephemeral`]
/// and therefore cannot be the record behind an external handoff. That is the
/// whole reason it exists as a named type rather than as a default: a caller
/// that reaches for it gets a refusal at the boundary instead of a green test
/// that means nothing.
#[derive(Debug, Clone)]
pub struct MemoryJournal {
    /// Every committed entry, in append order.
    committed: Vec<JournalEntry>,
    /// The last kind recorded per key, the append fence's index.
    ///
    /// Without it every append would walk the whole sequence to find the
    /// key's last event, and an append would cost the journal's own length:
    /// measured at 521 ns per append against an empty journal and 50,886 ns
    /// against 8,000 prior attempts, which is the O(n²) a long-lived bot
    /// would grind against. The index makes the ladder check constant.
    ladder: std::collections::HashMap<EffectKey, EventKind>,
    /// Latest durable outcome and its exact position per attempt.
    outcomes: std::collections::HashMap<EffectKey, (JournalPosition, EffectEvidence)>,
    /// Encoded frame bytes admitted, kept in step with `committed`.
    committed_bytes: u64,
}

impl Default for MemoryJournal {
    fn default() -> Self {
        Self::new()
    }
}

impl MemoryJournal {
    /// An empty journal at the genesis.
    #[must_use]
    pub fn new() -> Self {
        Self {
            committed: Vec::new(),
            ladder: std::collections::HashMap::new(),
            outcomes: std::collections::HashMap::new(),
            committed_bytes: 0,
        }
    }

    /// Every committed entry, in append order.
    #[must_use]
    pub fn committed(&self) -> &[JournalEntry] {
        &self.committed
    }

    /// The committed events, in append order.
    ///
    /// No `#[must_use]` here: the iterator this returns already carries one, so
    /// repeating it would be a second attribute saying the same thing.
    pub fn events(&self) -> impl Iterator<Item = &EffectEvent> {
        self.committed.iter().map(JournalEntry::event)
    }

    /// Recompute the chain and report the first disagreement, if any.
    ///
    /// # Errors
    ///
    /// [`ChainBreak`] when a recorded position does not follow from the events.
    pub fn verify(&self) -> Result<JournalPosition, ChainBreak> {
        verify_chain(&self.committed)
    }

    /// What the journal has learned about each attempt.
    #[must_use]
    pub fn recover(&self) -> Recovered {
        recover(self.events())
    }
}

impl EffectJournal for MemoryJournal {
    fn durability(&self) -> DurabilityPromise {
        DurabilityPromise::Ephemeral
    }

    fn tail(&self) -> JournalPosition {
        match self.committed.last() {
            Some(entry) => entry.position(),
            None => JournalPosition::genesis(),
        }
    }

    /// Copied out rather than borrowed, because the trait hands the caller a
    /// value it owns: `recover` folds over the sequence and a journal that
    /// handed back a borrow would tie the fold to the journal's own lifetime,
    /// which is exactly the coupling a caller reading a journal it is about to
    /// give back does not want.
    fn committed(&self) -> Result<Vec<EffectEvent>, JournalError> {
        Ok(self.events().copied().collect())
    }

    fn committed_entries(&self) -> Result<Vec<JournalEntry>, JournalError> {
        Ok(self.committed.clone())
    }

    fn committed_entry(
        &self,
        position: JournalPosition,
    ) -> Result<Option<JournalEntry>, JournalError> {
        let Some(index) = position
            .sequence()
            .checked_sub(1)
            .and_then(|n| usize::try_from(n).ok())
        else {
            return Ok(None);
        };
        Ok(self
            .committed
            .get(index)
            .copied()
            .filter(|entry| entry.position() == position))
    }

    fn outcome_at(
        &self,
        key: EffectKey,
    ) -> Result<Option<(JournalPosition, EffectEvidence)>, JournalError> {
        Ok(self.outcomes.get(&key).copied())
    }

    /// This journal's own counts against the shipped ceilings, reporting that it
    /// does not continue.
    ///
    /// Measured rather than zeroed: an in-memory journal refuses at exactly the
    /// same ceilings the file adapter does, so a caller that wants to know how
    /// close it is is asking a real question and gets a real number. The
    /// continuation answer is `false`, which is the whole reason an ephemeral
    /// run behaves exactly as it did before continuation existed.
    fn continuation_watermark(&self) -> Result<ContinuationWatermark, JournalError> {
        let events = u64::saturating_from(self.committed.len());
        let limit = u64::saturating_from(MAX_JOURNAL_EVENTS);
        Ok(ContinuationWatermark::inert(
            events,
            limit,
            self.committed_bytes,
            MAX_JOURNAL_BYTES,
        ))
    }

    fn compare_and_append(
        &mut self,
        expected_tail: JournalPosition,
        event: &EffectEvent,
    ) -> Result<DurableAck, JournalError> {
        let actual = self.tail();
        let key = event.key();
        check_append_order(expected_tail, actual, event, self.ladder.get(&key).copied())?;
        let requested_events = u64::saturating_from(self.committed.len()).saturating_add(1);
        let event_limit = u64::saturating_from(MAX_JOURNAL_EVENTS);
        if requested_events > event_limit {
            let refusal = Err(JournalError::CapacityExceeded {
                resource: JournalLimitKind::Events,
                limit: event_limit,
                requested: requested_events,
            });
            lgwks_std::trace::debug!(error = ?refusal.as_ref().err(), "compare_and_append: returning an error to the caller");
            return refusal;
        }
        let payload_len =
            u64::saturating_from(event.to_bytes().map_err(JournalError::Encoding)?.len());
        let requested_bytes = self
            .committed_bytes
            .saturating_add(payload_len)
            .saturating_add(36);
        if requested_bytes > MAX_JOURNAL_BYTES {
            let refusal = Err(JournalError::CapacityExceeded {
                resource: JournalLimitKind::Bytes,
                limit: MAX_JOURNAL_BYTES,
                requested: requested_bytes,
            });
            lgwks_std::trace::debug!(error = ?refusal.as_ref().err(), "compare_and_append: returning an error to the caller");
            return refusal;
        }
        let sequence = actual
            .sequence()
            .checked_add(1)
            .ok_or(JournalError::Exhausted)?;
        let position = JournalPosition {
            sequence,
            head: chain(actual, event)?,
        };
        self.committed.push(JournalEntry::new(position, *event));
        self.committed_bytes = requested_bytes;
        self.ladder.insert(key, event.kind());
        if let EffectEvent::OutcomeObserved { evidence, .. } = *event {
            self.outcomes.insert(key, (position, evidence));
        }
        Ok(DurableAck::new(position, self.durability()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::effect::{
        ActionDigest, ActionId, AttemptId, EnvironmentEpoch, EnvironmentId, FlowRevision, RunId,
    };

    const RUN: &str = "0102030405060708090a0b0c0d0e0f10";
    const ACTION: &str = "1112131415161718191a1b1c1d1e1f20";
    const OTHER_ACTION: &str = "3132333435363738393a3b3c3d3e3f40";
    const ENV: &str = "2122232425262728292a2b2c2d2e2f30";
    const FLOW_HEX: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
    const DIGEST_HEX: &str = "f0f1f2f3f4f5f6f7f8f9fafbfcfdfeffe0e1e2e3e4e5e6e7e8e9eaebecedeeef";
    const PREDICATE: &str = "4142434445464748494a4b4c4d4e4f50";

    /// The crate refuses a panicking path anywhere, tests included, so a test
    /// that needs a parsed value returns `Result` and propagates. A failure
    /// then names the cause instead of a line number in a macro.
    type TestResult = Result<(), Box<dyn std::error::Error>>;

    /// A key for the shared run and action, at the given attempt and epoch.
    fn key(attempt: &str, epoch: &str) -> Result<EffectKey, Box<dyn std::error::Error>> {
        build(ACTION, attempt, epoch)
    }

    /// A key for a different action, so identity comparisons have something to
    /// differ from.
    fn other_key(attempt: &str, epoch: &str) -> Result<EffectKey, Box<dyn std::error::Error>> {
        build(OTHER_ACTION, attempt, epoch)
    }

    /// Assemble a key from wire-form parts.
    fn build(
        action: &str,
        attempt: &str,
        epoch: &str,
    ) -> Result<EffectKey, Box<dyn std::error::Error>> {
        let key_run = RunId::from_hex(RUN)?;
        let key_action = ActionId::from_hex(action)?;
        let key_attempt = AttemptId::from_decimal(attempt)?;
        let key_flow_revision = FlowRevision::from_tagged("blake3_256", FLOW_HEX)?;
        let key_digest = ActionDigest::from_tagged("blake3_256", DIGEST_HEX)?;
        let key_environment = EnvironmentId::from_hex(ENV)?;
        let key_epoch = EnvironmentEpoch::from_decimal(epoch)?;
        Ok(
            crate::effect::EffectIdentity::new(key_run, key_environment, key_flow_revision).key(
                key_action,
                key_attempt,
                key_digest,
                key_epoch,
            ),
        )
    }

    /// A verification record naming a real predicate and a real observation
    /// digest.
    fn satisfied() -> Result<Verification, Box<dyn std::error::Error>> {
        Ok(Verification::new(
            Id128::from_hex(PREDICATE)?,
            1,
            blake3(b"observations"),
            VerificationResult::Satisfied,
        ))
    }

    /// Append `event` to `journal` at its own tail, which is what a correct
    /// caller does.
    fn append(
        journal: &mut MemoryJournal,
        event: EffectEvent,
    ) -> Result<DurableAck, Box<dyn std::error::Error>> {
        let tail = journal.tail();
        Ok(journal.compare_and_append(tail, &event)?)
    }

    #[test]
    fn memory_journal_refuses_history_beyond_its_declared_limit() -> TestResult {
        let mut journal = MemoryJournal::new();
        for attempt in 1_u64..=100_000 {
            let key = key(&attempt.to_string(), "1")?;
            journal.compare_and_append(journal.tail(), &EffectEvent::IntentAdmitted { key })?;
        }

        let extra_key = key("100001", "1")?;
        let extra = journal.compare_and_append(
            journal.tail(),
            &EffectEvent::IntentAdmitted { key: extra_key },
        );
        assert!(
            matches!(
                &extra,
                Err(JournalError::CapacityExceeded {
                    resource: JournalLimitKind::Events,
                    limit: 100_000,
                    requested: 100_001,
                })
            ),
            "the refusal names both the hard limit and requested size: {extra:?}"
        );
        assert_eq!(
            journal.committed().len(),
            100_000,
            "history must remain complete at the declared limit; extra append was {extra:?}"
        );
        Ok(())
    }

    /// Walk one attempt to the point where the irreversible boundary is next.
    fn admit_and_prepare(
        journal: &mut MemoryJournal,
        key: EffectKey,
    ) -> Result<(), Box<dyn std::error::Error>> {
        append(journal, EffectEvent::IntentAdmitted { key })?;
        append(journal, EffectEvent::DispatchPrepared { key })?;
        Ok(())
    }

    /// A journal holding one attempt walked past the boundary to an applied
    /// outcome: the state every verification test starts from.
    fn applied_journal(key: EffectKey) -> Result<MemoryJournal, Box<dyn std::error::Error>> {
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, key)?;
        append(
            &mut journal,
            EffectEvent::OutcomeObserved {
                key,
                evidence: EffectEvidence::Applied,
            },
        )?;
        Ok(journal)
    }

    #[test]
    fn genesis_is_not_a_zero_head() -> TestResult {
        let genesis = JournalPosition::genesis();
        assert_eq!(genesis.sequence(), 0);
        assert_ne!(genesis.head(), blake3(&[]));
        assert_eq!(genesis, JournalPosition::genesis());
        Ok(())
    }

    #[test]
    fn an_append_advances_the_sequence_and_moves_the_head() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        let genesis = journal.tail();
        let ack = append(&mut journal, EffectEvent::IntentAdmitted { key })?;
        assert_eq!(ack.position().sequence(), 1);
        assert_ne!(ack.position().head(), genesis.head());
        assert_eq!(journal.tail(), ack.position());
        Ok(())
    }

    #[test]
    fn the_acknowledgment_names_the_promise_that_was_claimed() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        let ack = append(&mut journal, EffectEvent::IntentAdmitted { key })?;
        assert_eq!(ack.promise(), DurabilityPromise::Ephemeral);
        Ok(())
    }

    #[test]
    fn positioned_readback_resolves_one_exact_sequence_and_head() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        let ack = append(&mut journal, EffectEvent::IntentAdmitted { key })?;
        let entries = EffectJournal::committed_entries(&journal)?;
        let expected = entries.first().copied().ok_or("committed entry missing")?;

        assert_eq!(
            journal.committed_entry(ack.position())?,
            Some(expected),
            "the acknowledged position returns its exact committed entry"
        );
        let wrong_head = JournalPosition {
            sequence: ack.position().sequence(),
            head: blake3(b"not the committed head"),
        };
        assert_eq!(
            journal.committed_entry(wrong_head)?,
            None,
            "a matching sequence cannot authorize a different history head"
        );
        Ok(())
    }

    #[test]
    fn two_events_in_another_order_are_a_different_head() -> TestResult {
        let first = key("1", "1")?;
        let second = other_key("1", "1")?;

        let mut forwards = MemoryJournal::new();
        append(&mut forwards, EffectEvent::IntentAdmitted { key: first })?;
        append(&mut forwards, EffectEvent::IntentAdmitted { key: second })?;

        let mut backwards = MemoryJournal::new();
        append(&mut backwards, EffectEvent::IntentAdmitted { key: second })?;
        append(&mut backwards, EffectEvent::IntentAdmitted { key: first })?;

        assert_eq!(forwards.tail().sequence(), backwards.tail().sequence());
        assert_ne!(forwards.tail().head(), backwards.tail().head());
        Ok(())
    }

    #[test]
    fn a_stale_tail_is_refused_and_changes_nothing() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        let genesis = journal.tail();
        let committed = append(&mut journal, EffectEvent::IntentAdmitted { key })?;

        let refused = journal.compare_and_append(genesis, &EffectEvent::DispatchPrepared { key });
        match refused {
            Err(JournalError::TailMismatch { expected, actual }) => {
                assert_eq!(expected, genesis);
                assert_eq!(actual, committed.position());
            }
            other => return Err(format!("expected a tail mismatch, got {other:?}").into()),
        }
        assert_eq!(journal.committed().len(), 1);
        assert_eq!(journal.tail(), committed.position());
        Ok(())
    }

    #[test]
    fn a_second_controller_reading_the_same_tail_is_fenced() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        let both_saw = journal.tail();

        let winner = journal.compare_and_append(both_saw, &EffectEvent::IntentAdmitted { key })?;

        let loser = journal.compare_and_append(both_saw, &EffectEvent::DispatchPrepared { key });
        assert!(
            matches!(loser, Err(JournalError::TailMismatch { .. })),
            "the second writer from a stale reading must be refused"
        );
        assert_eq!(journal.committed().len(), 1);
        assert_eq!(journal.tail(), winner.position());
        Ok(())
    }

    #[test]
    fn a_dispatch_before_its_intent_is_refused() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        let refused =
            journal.compare_and_append(journal.tail(), &EffectEvent::DispatchPrepared { key });
        match refused {
            Err(JournalError::OutOfOrder {
                key: refused_key,
                expected,
                attempted,
            }) => {
                assert_eq!(*refused_key, key);
                assert_eq!(expected, Some(EventKind::IntentAdmitted));
                assert_eq!(attempted, EventKind::DispatchPrepared);
            }
            other => return Err(format!("expected an ordering refusal, got {other:?}").into()),
        }
        assert!(journal.committed().is_empty());
        Ok(())
    }

    #[test]
    fn the_ladder_is_walked_once_per_key() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        append(&mut journal, EffectEvent::IntentAdmitted { key })?;
        let refused =
            journal.compare_and_append(journal.tail(), &EffectEvent::IntentAdmitted { key });
        assert!(
            matches!(
                refused,
                Err(JournalError::OutOfOrder {
                    expected: Some(EventKind::DispatchPrepared),
                    attempted: EventKind::IntentAdmitted,
                    ..
                })
            ),
            "a second intent for one key is how a blind resend becomes representable"
        );
        assert_eq!(journal.committed().len(), 1);
        Ok(())
    }

    #[test]
    fn a_verified_attempt_refuses_everything_but_a_later_verification() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = applied_journal(key)?;
        append(
            &mut journal,
            EffectEvent::Verified {
                key,
                verification: satisfied()?,
            },
        )?;
        let refused = journal.compare_and_append(
            journal.tail(),
            &EffectEvent::OutcomeObserved {
                key,
                evidence: EffectEvidence::NotApplied,
            },
        );
        assert!(
            matches!(
                refused,
                Err(JournalError::OutOfOrder {
                    expected: Some(EventKind::Verified),
                    attempted: EventKind::OutcomeObserved,
                    ..
                })
            ),
            "only a later verification follows a verification; the effect is not re-settled"
        );
        Ok(())
    }

    #[test]
    fn an_ephemeral_journal_reports_itself_and_refuses_the_boundary() -> TestResult {
        let journal = MemoryJournal::new();
        assert_eq!(journal.durability(), DurabilityPromise::Ephemeral);
        match journal.admit_external_handoff() {
            Err(JournalError::PromiseUnmet { required, offered }) => {
                assert_eq!(required, DurabilityPromise::ProcessCrash);
                assert_eq!(offered, DurabilityPromise::Ephemeral);
            }
            other => {
                return Err(format!(
                    "an in-memory journal must not host an external handoff, got {other:?}"
                )
                .into());
            }
        }
        Ok(())
    }

    #[test]
    fn only_a_crash_surviving_promise_clears_the_boundary() {
        assert!(!DurabilityPromise::Ephemeral.survives_process_crash());
        assert!(DurabilityPromise::ProcessCrash.survives_process_crash());
        assert!(DurabilityPromise::PowerLoss.survives_process_crash());
        assert!(DurabilityPromise::PowerLoss > DurabilityPromise::ProcessCrash);
    }

    #[test]
    fn a_prepared_dispatch_with_no_result_recovers_unknown() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, key)?;

        let recovered = journal.recover();
        assert_eq!(recovered.len(), 1);
        assert_eq!(recovered.status(key), Some(AttemptStatus::OutcomeUnknown));
        assert_eq!(recovered.uncertain(), vec![key]);
        Ok(())
    }

    #[test]
    fn an_observed_not_applied_settles_the_unknown() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, key)?;
        append(
            &mut journal,
            EffectEvent::OutcomeObserved {
                key,
                evidence: EffectEvidence::NotApplied,
            },
        )?;

        let recovered = journal.recover();
        assert_eq!(recovered.status(key), Some(AttemptStatus::NotApplied));
        assert!(
            recovered.uncertain().is_empty(),
            "evidence that the effect did not land is what ends the uncertainty"
        );
        Ok(())
    }

    #[test]
    fn an_admitted_intent_alone_is_not_uncertain() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        append(&mut journal, EffectEvent::IntentAdmitted { key })?;
        assert_eq!(journal.recover().status(key), Some(AttemptStatus::Prepared));
        assert!(journal.recover().uncertain().is_empty());
        Ok(())
    }

    #[test]
    fn a_satisfied_predicate_is_the_only_verified_state() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = applied_journal(key)?;
        append(
            &mut journal,
            EffectEvent::Verified {
                key,
                verification: satisfied()?,
            },
        )?;
        assert_eq!(journal.recover().status(key), Some(AttemptStatus::Verified));
        Ok(())
    }

    #[test]
    fn a_predicate_that_did_not_hold_is_not_a_verification() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = applied_journal(key)?;
        append(
            &mut journal,
            EffectEvent::Verified {
                key,
                verification: Verification::new(
                    Id128::from_hex(PREDICATE)?,
                    1,
                    blake3(b"observations"),
                    VerificationResult::NotSatisfied,
                ),
            },
        )?;

        let status = journal.recover().status(key);
        assert_eq!(status, Some(AttemptStatus::VerificationFailed));
        assert_ne!(status, Some(AttemptStatus::Applied));
        assert_ne!(status, Some(AttemptStatus::Verified));
        assert!(
            !status.is_some_and(AttemptStatus::is_uncertain),
            "the effect is known to have landed, so a failed predicate is not uncertainty"
        );
        Ok(())
    }

    fn verdict(
        version: u64,
        result: VerificationResult,
    ) -> Result<Verification, Box<dyn std::error::Error>> {
        Ok(Verification::new(
            Id128::from_hex(PREDICATE)?,
            version,
            blake3(format!("observations at version {version}").as_bytes()),
            result,
        ))
    }

    /// #260: a verdict is revisable. `Verified`, then the world moves and the
    /// predicate no longer holds, then it holds again: each is recordable, the
    /// recovered status follows the latest, and the verdict names the predicate
    /// version it was decided at.
    #[test]
    fn a_verification_can_be_revised_in_both_directions() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = applied_journal(key)?;
        for (version, result) in [
            (1, VerificationResult::Satisfied),
            (2, VerificationResult::NotSatisfied),
            (3, VerificationResult::Satisfied),
        ] {
            append(
                &mut journal,
                EffectEvent::Verified {
                    key,
                    verification: verdict(version, result)?,
                },
            )?;
        }

        let recovered = journal.recover();
        assert_eq!(recovered.status(key), Some(AttemptStatus::Verified));
        assert_eq!(
            recovered
                .verification(key)
                .map(|found| found.predicate_version()),
            Some(3),
            "the status is qualified by the version it was decided at"
        );
        let trail: Vec<_> = recovered
            .history(key)
            .iter()
            .map(|change| (change.from(), change.to(), change.position()))
            .collect();
        assert_eq!(
            trail,
            vec![
                (None, AttemptStatus::Prepared, 0),
                (
                    Some(AttemptStatus::Prepared),
                    AttemptStatus::OutcomeUnknown,
                    1
                ),
                (
                    Some(AttemptStatus::OutcomeUnknown),
                    AttemptStatus::Applied,
                    2
                ),
                (Some(AttemptStatus::Applied), AttemptStatus::Verified, 3),
                (
                    Some(AttemptStatus::Verified),
                    AttemptStatus::VerificationFailed,
                    4
                ),
                (
                    Some(AttemptStatus::VerificationFailed),
                    AttemptStatus::Verified,
                    5
                ),
            ],
            "what changed, what it superseded, and where in the journal"
        );
        Ok(())
    }

    #[test]
    fn recovery_lists_attempts_in_admission_order() -> TestResult {
        let first = key("1", "1")?;
        let second = other_key("2", "1")?;
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, second)?;
        admit_and_prepare(&mut journal, first)?;

        let attempts = journal.recover().attempts().to_vec();
        assert_eq!(attempts.len(), 2);
        assert_eq!(attempts[0].key(), second);
        assert_eq!(attempts[1].key(), first);
        Ok(())
    }

    #[test]
    fn verify_accepts_an_untampered_chain() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, key)?;
        assert_eq!(journal.verify()?, journal.tail());
        Ok(())
    }

    #[test]
    fn verify_rejects_a_rewritten_position() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, key)?;

        let mut tampered = journal.committed().to_vec();
        let last = tampered
            .pop()
            .ok_or("the journal should hold two entries after admit and prepare")?;
        tampered.push(JournalEntry::new(JournalPosition::genesis(), *last.event()));

        match verify_chain(&tampered) {
            Err(broken) => assert_eq!(broken.at(), 2),
            Ok(position) => {
                return Err(
                    format!("a rewritten position must be detected, got {position}").into(),
                );
            }
        }
        Ok(())
    }

    #[test]
    fn verify_rejects_a_reordered_chain() -> TestResult {
        let key = key("1", "1")?;
        let mut journal = MemoryJournal::new();
        admit_and_prepare(&mut journal, key)?;

        let mut reordered = journal.committed().to_vec();
        reordered.reverse();

        match verify_chain(&reordered) {
            Err(broken) => assert_eq!(broken.at(), 1),
            Ok(position) => {
                return Err(format!("a reordered chain must be detected, got {position}").into());
            }
        }
        Ok(())
    }

    #[test]
    fn a_key_encodes_deterministically() -> TestResult {
        let key = key("1", "1")?;
        let bytes = key.to_bytes()?;
        assert_eq!(
            bytes.as_slice(),
            key.to_bytes()?.as_slice(),
            "one key encodes the same way twice"
        );
        assert!(
            !bytes.is_empty(),
            "and the encoding carries something: {bytes:?}"
        );
        Ok(())
    }

    #[test]
    fn keys_differing_in_one_field_differ_in_bytes() -> TestResult {
        let base = key("1", "1")?;
        let later_attempt = key("2", "1")?;
        let later_epoch = key("1", "2")?;
        let other = other_key("1", "1")?;

        assert_ne!(
            base.to_bytes()?.as_slice(),
            later_attempt.to_bytes()?.as_slice()
        );
        assert_ne!(
            base.to_bytes()?.as_slice(),
            later_epoch.to_bytes()?.as_slice()
        );
        assert_ne!(base.to_bytes()?.as_slice(), other.to_bytes()?.as_slice());
        Ok(())
    }

    #[test]
    fn an_event_encoding_carries_its_kind_and_its_key() -> TestResult {
        let key = key("1", "1")?;
        let admitted = EffectEvent::IntentAdmitted { key }.to_bytes()?;
        let prepared = EffectEvent::DispatchPrepared { key }.to_bytes()?;
        let applied = EffectEvent::OutcomeObserved {
            key,
            evidence: EffectEvidence::Applied,
        }
        .to_bytes()?;
        let not_applied = EffectEvent::OutcomeObserved {
            key,
            evidence: EffectEvidence::NotApplied,
        }
        .to_bytes()?;

        assert_ne!(admitted.as_slice(), prepared.as_slice());
        assert_ne!(applied.as_slice(), not_applied.as_slice());

        // What the encoding says about the event is read back out of it. A
        // field that failed to encode would not shorten the buffer, since the
        // archived form is one width for every variant of a type, so comparing
        // lengths would not have caught it. Decoding does.
        for (bytes, event) in [
            (admitted, EffectEvent::IntentAdmitted { key }),
            (prepared, EffectEvent::DispatchPrepared { key }),
            (
                applied,
                EffectEvent::OutcomeObserved {
                    key,
                    evidence: EffectEvidence::Applied,
                },
            ),
            (
                not_applied,
                EffectEvent::OutcomeObserved {
                    key,
                    evidence: EffectEvidence::NotApplied,
                },
            ),
        ] {
            assert_eq!(
                lgwks_std::wire::from_bytes::<EffectEvent, WireError>(&bytes)?,
                event
            );
        }
        Ok(())
    }

    #[test]
    fn a_verification_encoding_names_its_predicate_and_its_version() -> TestResult {
        let key = key("1", "1")?;
        let first = EffectEvent::Verified {
            key,
            verification: satisfied()?,
        }
        .to_bytes()?;
        let next_version = EffectEvent::Verified {
            key,
            verification: Verification::new(
                Id128::from_hex(PREDICATE)?,
                2,
                blake3(b"observations"),
                VerificationResult::Satisfied,
            ),
        }
        .to_bytes()?;
        assert_ne!(first.as_slice(), next_version.as_slice());
        assert_eq!(
            lgwks_std::wire::from_bytes::<EffectEvent, WireError>(&first)?,
            EffectEvent::Verified {
                key,
                verification: satisfied()?,
            }
        );
        assert_eq!(
            lgwks_std::wire::from_bytes::<EffectEvent, WireError>(&next_version)?,
            EffectEvent::Verified {
                key,
                verification: Verification::new(
                    Id128::from_hex(PREDICATE)?,
                    2,
                    blake3(b"observations"),
                    VerificationResult::Satisfied,
                ),
            }
        );
        Ok(())
    }
}