polyc-eventlog 2026.9.0

//! Append-only conversation event log on a `commonware-storage` journal.
//!
//! This crate persists the ordered stream of events that make up a conversation
//! (user messages, planner decisions, tool calls, …) to an append-only log
//! backed by the Commonware storage stack — keeping persistence on the
//! Commonware primitives rather than a relational store.
//!
//! # Storage primitive
//!
//! [`EventLog`] wraps
//! [`commonware_storage::journal::contiguous::variable::Journal`]: a
//! **contiguous, position-based, variable-length** append-only journal. It is
//! the natural fit here:
//!
//! - **Append-only.** [`EventLog::append`] writes one [`Event`] and returns the
//!   monotonically increasing `u64` *position* the journal assigned it.
//!   Positions start at `0` and never reused; pruning earlier entries does not
//!   shift later positions.
//! - **Ordered replay.** [`EventLog::replay`] returns every event in append
//!   order, each paired with its position. **Append order is the ordering
//!   contract**: the caller appends events in conversation order (turn, then
//!   sequence within a turn), and replay yields them back in exactly that
//!   order. The position therefore *is* the (turn, seq) ordinal flattened into
//!   one strictly increasing sequence — there is no separate sort key to
//!   maintain, which is precisely what an append-only log buys us.
//! - **Variable-length items.** Each event's `payload` is an opaque,
//!   buffa-encoded byte blob of arbitrary size; the `variable` journal stores
//!   variable-length items natively (the `contiguous::fixed` sibling is for
//!   fixed-width records and would not fit).
//!
//! # Runtime genericity (tokio vs. deterministic)
//!
//! The journal — and therefore [`EventLog`] — is generic over a
//! [`commonware_storage::Context`] (the `Storage + Clock + Metrics` bound
//! every Commonware storage type carries). Production drives it on the
//! `commonware_runtime::tokio` backend; tests drive it on the
//! `commonware_runtime::deterministic` backend for seeded, reproducible runs.
//! The two never nest: the Commonware runtime cannot be started from inside a
//! live tokio runtime, so every tokio process that embeds this crate runs it
//! on a dedicated thread. The state plane does so for the conversation journal
//! it alone writes; the control plane does so for the non-conversation logs it
//! keeps of its own. This crate stays runtime-agnostic and leaves that hosting
//! decision to the caller.
//!
//! # Conversation scoping
//!
//! One [`EventLog`] instance maps to one conversation's log, identified by the
//! storage *partition* name passed to [`EventLog::open`] (derive it from the
//! conversation id, e.g. `format!("conv-{uid}")`). Distinct conversations use
//! distinct partitions and so are fully isolated on disk.
//!
//! # Example
//!
//! <!--
//! Marked `ignore`, not run: a runnable doctest statically links the entire
//! Commonware storage stack into its own dedicated binary, and that link
//! OOMs/bus-errors CI's linker. The example is mirrored verbatim by the
//! `doc_example_open_append_replay` unit test, which folds into the crate's
//! existing (already-linked) test binary rather than adding a second heavy
//! link — so the snippet stays verified without the extra link unit.
//! -->
//! ```ignore
//! use commonware_runtime::{deterministic, Runner};
//! use polyc_eventlog::{Event, EventLog, EventLogConfig};
//!
//! let executor = deterministic::Runner::default();
//! executor.start(|context| async move {
//!     let log = EventLog::open(context, EventLogConfig::for_partition("conv-1"))
//!         .await
//!         .expect("open log");
//!
//!     log.append(&Event::new("user_msg", b"hello".to_vec())).await.unwrap();
//!     log.append(&Event::new("tool_call", b"\x01\x02".to_vec())).await.unwrap();
//!     log.commit().await.unwrap();
//!
//!     let events = log.replay().await.unwrap();
//!     assert_eq!(events.len(), 2);
//!     assert_eq!(events[0].kind, "user_msg");
//! });
//! ```

pub mod checkpoint;
pub mod error;
mod metrics;

pub use checkpoint::EventCountCheckpoint;
pub use error::EventLogError;
pub use polyc_eventlog_model::integrity;
pub use polyc_eventlog_model::integrity::{
    IntegrityError, MMR_SIGNED_ROOT_KIND, RootStanding, extend_and_sign, rebuild_from_events,
    root_standing_with_trust, verify_extension_with_trust, verify_replay, verify_replay_with_trust,
};
pub use polyc_eventlog_model::nav;
pub use polyc_eventlog_model::taint;
pub use polyc_eventlog_model::taint::{
    GrantedCapabilities, TrifectaLegs, TrustTag, any_untrusted, any_untrusted_excluding,
    trifecta_legs,
};
pub use polyc_eventlog_model::{BoundedReplay, Event, EventCfg};

/// Force-register this crate's Prometheus append-latency histogram.
///
/// Makes it appear in a `/metrics` scrape immediately — before any event has
/// been appended. Idempotent (backed by a `OnceLock`); call once at process
/// startup, alongside any other crate's own `init_metrics`.
pub fn init_metrics() {
    metrics::force();
}

use commonware_runtime::buffer::paged::CacheRef;
use commonware_storage::journal::contiguous::{Contiguous as _, variable};
use commonware_utils::sync::AsyncMutex;
use commonware_utils::{NZU16, NZU64, NZUsize};
use futures::StreamExt as _;
use std::num::{NonZeroU16, NonZeroU64, NonZeroUsize};

/// Buffer size (in items) for the replay stream from the underlying journal.
const REPLAY_BUFFER: NonZeroUsize = NZUsize!(1024);

/// One event [`EventLog::replay_quarantining`] could not decode, and why.
#[derive(Debug, Clone)]
pub struct QuarantinedItem {
    /// Journal position of the corrupted item.
    pub position: u64,
    /// The underlying decode/storage error's `Display` text.
    pub error: String,
}

/// Configuration for opening an [`EventLog`].
///
/// Most fields mirror the underlying journal's tuning knobs and have sensible
/// defaults via [`EventLogConfig::for_partition`]; only the `partition`
/// (which conversation's log) is mandatory.
#[derive(Debug, Clone)]
pub struct EventLogConfig {
    /// Storage partition name — one per conversation. Sub-partitions for the
    /// data and offset indexes are derived from it by the journal.
    pub partition: String,

    /// Number of events stored per journal section. Sections roll over at this
    /// count; only the final (partial) section is replayed on open to recover
    /// the exact size. **Immutable once a partition exists** — changing it
    /// across restarts corrupts the log.
    pub items_per_section: NonZeroU64,

    /// Decode-time bounds applied to each event during [`EventLog::replay`].
    pub event_cfg: EventCfg,

    /// Page size for the read cache over the underlying storage blobs.
    pub page_size: NonZeroU16,

    /// Page cache capacity, in pages.
    pub page_cache_pages: NonZeroUsize,

    /// Per-section write buffer size, in bytes.
    pub write_buffer: NonZeroUsize,
}

impl EventLogConfig {
    /// Build a config for `partition` with defaults for every other field.
    ///
    /// Defaults: 1024 events per section, [`EventCfg::DEFAULT`] decode bounds,
    /// a 16 KiB page size with a 64-page cache, and a 64 KiB write buffer.
    #[must_use]
    pub fn for_partition(partition: impl Into<String>) -> Self {
        Self {
            partition: partition.into(),
            items_per_section: NZU64!(1024),
            event_cfg: EventCfg::DEFAULT,
            page_size: NZU16!(16384),
            page_cache_pages: NZUsize!(64),
            write_buffer: NZUsize!(65536),
        }
    }
}

/// An append-only, ordered log of conversation [`Event`]s.
///
/// Generic over a [`commonware_storage::Context`] so the same code runs on the
/// tokio backend in production and the deterministic backend in tests. See the
/// crate-level docs for the ordering contract and runtime-coexistence notes.
pub struct EventLog<E>
where
    E: commonware_storage::Context + commonware_runtime::BufferPooler,
{
    /// What this log was opened with, kept so [`EventLog::reset`] rebuilds the
    /// journal through the same authoritative builder [`EventLog::open`] used
    /// ([`journal_config`]).
    ///
    /// A reset that took its own configuration could reset a partition to a
    /// different section size, codec bound, or page cache than the one the
    /// partition was written with. Holding the opened configuration removes
    /// that possibility rather than documenting against it.
    config: EventLogConfig,

    /// The journal's append/commit/sync/snapshot operations take `&mut self`
    /// (commonware 2026.7's journal API), but `EventLog` hands out a single
    /// shared handle to a control-plane host, forensics reads, and the
    /// workqueue alike. This lock recovers that shared surface; it is not a
    /// new concurrency model — a `Lease` at a higher layer already serializes
    /// writers per conversation, so contention here is only ever between the
    /// lone writer and concurrent readers.
    journal: AsyncMutex<variable::Journal<E, Event>>,
}

/// The one place an [`EventLogConfig`] becomes a Commonware journal
/// configuration.
///
/// Both [`EventLog::open`] and [`EventLog::reset`] go through it, so a reset
/// cannot drift from the open it replaces: the partition name, section size,
/// codec bounds, page cache, and write buffer are decided once, here.
fn journal_config<E>(context: &E, config: &EventLogConfig) -> variable::Config<EventCfg>
where
    E: commonware_storage::Context + commonware_runtime::BufferPooler,
{
    variable::Config {
        partition: config.partition.clone(),
        items_per_section: config.items_per_section,
        compression: None,
        codec_config: config.event_cfg,
        page_cache: CacheRef::from_pooler(context, config.page_size, config.page_cache_pages),
        write_buffer: config.write_buffer,
    }
}

impl<E> EventLog<E>
where
    E: commonware_storage::Context + commonware_runtime::BufferPooler,
{
    /// Open (creating if absent, recovering if present) the event log for a
    /// conversation on the given runtime `context`.
    ///
    /// On open the journal replays only its final section to recover the exact
    /// append size, and self-heals any data/offset divergence left by a crash.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the underlying storage fails to
    /// initialize or recover the journal.
    pub async fn open(context: E, config: EventLogConfig) -> Result<Self, EventLogError> {
        let journal_cfg = journal_config(&context, &config);
        let journal = variable::Journal::init(context, journal_cfg).await?;
        Ok(Self {
            config,
            journal: AsyncMutex::new(journal),
        })
    }

    /// Empty the partition recoverably: consume the handle, reset the journal
    /// to logical position zero, and return the healthy journal that replaces
    /// it.
    ///
    /// This is the durable half of erasing, rewriting, repairing, and
    /// migrating a partition. It delegates to Commonware's
    /// [`variable::Journal::init_at_size`], which stages the offsets reset
    /// intent durably BEFORE the data partition is cleared, so a crash at any
    /// point leaves a staged clear that the next
    /// [`variable::Journal::init`] finishes. Content can therefore never
    /// outlive the reset, and no interrupted reset can serve a short history.
    ///
    /// Commonware's `Mutable::destroy` cannot do this job: its own contract
    /// calls it final teardown that "is not crash-safe", and says reopening
    /// interrupted storage "may observe partially removed state". Use
    /// [`EventLog::reclaim`] for the physical removal that genuinely is final,
    /// and only after this reset has proven the journal empty.
    ///
    /// Deliberately consuming, in both directions. The old journal is dropped
    /// before the reset runs, so no handle survives the storage it described;
    /// and the caller gets the new journal back rather than reopening one, so
    /// nothing has to remember to.
    ///
    /// `context` is supplied rather than held because a Commonware context is
    /// not `Clone` — it is a supervision-tree identity. Pass the same
    /// `context.child(label)` the original [`EventLog::open`] was given, so
    /// the reset journal registers its metrics under the same node.
    ///
    /// The configuration is the one this log was opened with, held privately
    /// since [`EventLog::open`], so a reset cannot silently move a partition
    /// to different section, codec, or cache bounds than it was written with.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the reset cannot be staged or
    /// completed. The old handle is already gone when that happens, which is
    /// what Commonware's rule for a failed mutable operation requires.
    pub async fn reset(self, context: E) -> Result<Self, EventLogError> {
        let Self { config, journal } = self;
        // Before the reset, not after it. The old journal owns open blobs over
        // the storage `init_at_size` is about to clear, and Commonware no
        // longer vouches for a handle whose storage moved underneath it.
        drop(journal.into_inner());
        let journal_cfg = journal_config(&context, &config);
        let journal = variable::Journal::init_at_size(context, journal_cfg, 0).await?;
        Ok(Self {
            config,
            journal: AsyncMutex::new(journal),
        })
    }

    /// Physically remove an ALREADY-EMPTY partition's storage, consuming the
    /// handle. The second half of the erasure primitive (`#216`).
    ///
    /// This is Commonware's final-teardown `Mutable::destroy`, which is not
    /// crash-safe. It is used here only where crash-safety is no longer at
    /// stake: the caller must have run [`EventLog::reset`] first, and this
    /// refuses outright if the journal still holds events. An interrupted
    /// removal can therefore leave only absence or an empty remnant — never a
    /// short or partial history, which is the failure `reset` exists to
    /// prevent.
    ///
    /// Commonware's `destroy` is what removes the storage rather than a
    /// directory sweep of our own because the data and offsets partition names
    /// are private to Commonware's journal configuration. Naming them here
    /// would duplicate a layout this crate does not own.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::ReclaimNotEmpty`] if the journal still holds
    /// events, and [`EventLogError::Journal`] if the removal itself fails.
    pub async fn reclaim(self) -> Result<(), EventLogError> {
        let events = self.len().await;
        if events != 0 {
            return Err(EventLogError::ReclaimNotEmpty { events });
        }
        Ok(self.journal.into_inner().destroy().await?)
    }

    /// Append a single event, returning the position the journal assigned it.
    ///
    /// Positions are strictly increasing from `0` and define replay order. The
    /// caller must append in conversation order (turn, then seq within a turn)
    /// for replay to reflect that order.
    ///
    /// Takes `&self`: [`EventLog`]'s own lock (see the struct-level doc)
    /// recovers a shared reference over the journal's `&mut self` ops.
    ///
    /// Appends are buffered for durability; call [`EventLog::commit`] (or
    /// [`EventLog::sync`]) to guarantee they survive a crash.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the item cannot be encoded or the
    /// underlying storage write fails.
    pub async fn append(&self, event: &Event) -> Result<u64, EventLogError> {
        let start = std::time::Instant::now(); // determinism-allow: metrics-only timing, never persisted or replayed
        let result = self.journal.lock().await.append(event).await;
        metrics::record_append(result.is_ok(), start.elapsed());
        Ok(result?)
    }

    /// Number of events appended to the log (the position the *next* append
    /// will receive). Not reduced by pruning.
    pub async fn len(&self) -> u64 {
        self.journal.lock().await.size()
    }

    /// Whether the log has no appended events.
    pub async fn is_empty(&self) -> bool {
        self.len().await == 0
    }

    /// Replay every event in append order, each paired with its position.
    ///
    /// The returned `Vec` is ordered by position ascending (`0, 1, 2, …`),
    /// which is conversation order. Each tuple is `(position, event)`.
    ///
    /// This collects the full log into memory; it is intended for rebuilding
    /// in-memory conversation state on resume. For very large logs a streaming
    /// variant could be added later (the journal exposes a `Stream`), but the
    /// foundational API materializes for simplicity.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the journal cannot start the replay
    /// stream or if decoding any stored event fails.
    // `snapshot()` returns an owned, `'static` reader (unlike the borrowed
    // `reader()` of commonware 2026.5), so the journal lock is released as
    // soon as the snapshot is taken — the stream below never holds it.
    pub async fn replay_with_positions(&self) -> Result<Vec<(u64, Event)>, EventLogError> {
        let reader = self.journal.lock().await.snapshot().await?;
        let start = reader.bounds().start;
        let stream = reader.replay(start, REPLAY_BUFFER).await?;
        futures::pin_mut!(stream);
        let mut out = Vec::new();
        while let Some(item) = stream.next().await {
            out.push(item?);
        }
        Ok(out)
    }

    /// Replay every event in append order, each paired with its position —
    /// same as [`EventLog::replay_with_positions`] — but STOP pulling from
    /// the underlying replay stream the instant the cumulative payload bytes
    /// read so far exceed `max_bytes`.
    ///
    /// This is issue #1541's early-abort primitive: [`EventLog::replay_with_positions`]
    /// always drains the whole stream into one `Vec` before any caller can
    /// check its size, so a budget checked only after that call returns has
    /// already paid the full allocation cost it meant to avoid. This method
    /// instead checks the running byte total INSIDE the same loop that pulls
    /// from the stream, so the returned `Vec` never grows past `max_bytes`
    /// plus one event's own payload size (the one event whose read tips the
    /// budget over is kept, then the loop breaks — the rest of the
    /// partition, however large, is never fetched from the journal).
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the journal cannot start the
    /// replay stream or if decoding any stored item fails before the budget
    /// trips.
    pub async fn replay_with_positions_bounded(
        &self,
        max_bytes: u64,
    ) -> Result<BoundedReplay, EventLogError> {
        let reader = self.journal.lock().await.snapshot().await?;
        let start = reader.bounds().start;
        let stream = reader.replay(start, REPLAY_BUFFER).await?;
        futures::pin_mut!(stream);
        let mut events = Vec::new();
        let mut bytes_read: u64 = 0;
        let mut budget_exceeded = false;
        while let Some(item) = stream.next().await {
            let (position, event) = item?;
            bytes_read = bytes_read.saturating_add(event.payload.len() as u64);
            events.push((position, event));
            if bytes_read > max_bytes {
                budget_exceeded = true;
                break;
            }
        }
        Ok(BoundedReplay {
            events,
            bytes_read,
            budget_exceeded,
        })
    }

    /// Replay events in append order starting at position `start`, each paired
    /// with its position.
    ///
    /// The journal is position-indexed, so resuming at an offset is cheap — the
    /// reader seeks to `start` rather than scanning from zero. This is what lets
    /// a caller replay only the tail since a durable checkpoint instead of
    /// re-reading the whole partition every time. `start` is clamped up to the
    /// pruning boundary, and a `start` at or past the end yields an empty `Vec`.
    /// Returned tuples are `(position, event)` for positions in
    /// `[max(start, bounds.start), len)`, ascending.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the journal cannot start the replay
    /// stream or if decoding any stored event fails.
    // See [`EventLog::replay_with_positions`]: the snapshot is owned, so the
    // journal lock is released before the stream is consumed.
    pub async fn replay_from_with_positions(
        &self,
        start: u64,
    ) -> Result<Vec<(u64, Event)>, EventLogError> {
        let reader = self.journal.lock().await.snapshot().await?;
        let bounds = reader.bounds();
        let from = start.max(bounds.start);
        if from >= bounds.end {
            return Ok(Vec::new());
        }
        let stream = reader.replay(from, REPLAY_BUFFER).await?;
        futures::pin_mut!(stream);
        let mut out = Vec::new();
        while let Some(item) = stream.next().await {
            out.push(item?);
        }
        Ok(out)
    }

    /// Replay events in append order starting at position `start`, each
    /// paired with its position — same resume semantics as
    /// [`EventLog::replay_from_with_positions`] — but STOP pulling from the
    /// underlying replay stream the instant the cumulative payload bytes read
    /// so far exceed `max_bytes`, the same early-abort mechanic
    /// [`EventLog::replay_with_positions_bounded`] applies to a replay from
    /// the very start.
    ///
    /// This is the combined primitive neither of the other two replay
    /// methods can express alone: [`EventLog::replay_with_positions_bounded`]
    /// bounds bytes but always starts at position `0`, and
    /// [`EventLog::replay_from_with_positions`] resumes at `start` but has no
    /// byte cap, so a partition whose TAIL (the part after a caller-held
    /// watermark) is itself large could still be materialized in full before
    /// any caller ever gets a chance to reject it. This method closes that
    /// gap: a caller resuming from its own cached watermark (`polyc_query`'s
    /// per-partition decode cache is the first consumer) gets the same
    /// mid-stream budget enforcement a fresh replay already had, without
    /// paying to re-read anything before `start`.
    ///
    /// `start` is clamped up to the partition's own pruning boundary exactly
    /// as [`EventLog::replay_from_with_positions`] does, and a `start` at or
    /// past the partition's end yields an empty, `budget_exceeded: false`
    /// [`BoundedReplay`] rather than an error.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the journal cannot start the
    /// replay stream or if decoding any stored item fails before the budget
    /// trips.
    // See [`EventLog::replay_with_positions`]: the snapshot is owned, so the
    // journal lock is released before the stream is consumed.
    pub async fn replay_from_with_positions_bounded(
        &self,
        start: u64,
        max_bytes: u64,
    ) -> Result<BoundedReplay, EventLogError> {
        let reader = self.journal.lock().await.snapshot().await?;
        let bounds = reader.bounds();
        let from = start.max(bounds.start);
        if from >= bounds.end {
            return Ok(BoundedReplay {
                events: Vec::new(),
                bytes_read: 0,
                budget_exceeded: false,
            });
        }
        let stream = reader.replay(from, REPLAY_BUFFER).await?;
        futures::pin_mut!(stream);
        let mut events = Vec::new();
        let mut bytes_read: u64 = 0;
        let mut budget_exceeded = false;
        while let Some(item) = stream.next().await {
            let (position, event) = item?;
            bytes_read = bytes_read.saturating_add(event.payload.len() as u64);
            events.push((position, event));
            if bytes_read > max_bytes {
                budget_exceeded = true;
                break;
            }
        }
        Ok(BoundedReplay {
            events,
            bytes_read,
            budget_exceeded,
        })
    }

    /// Replay events in `[start, end)` — end EXCLUSIVE — each paired with its
    /// position, under the same byte cap
    /// [`EventLog::replay_from_with_positions_bounded`] applies.
    ///
    /// The only replay primitive here that accepts an UPPER bound. Every
    /// other one drains to the journal's tail: `replay_with_positions` and
    /// `replay_from_with_positions` have no cap at all, and the two
    /// `_bounded` siblings cap BYTES, which stops a large read but cannot
    /// express "these events and no others". A caller that knows the exact
    /// span it wants — one turn's events, say, located by a prior index —
    /// otherwise has to replay from `start` to the tail and discard the
    /// remainder, so the cost of fetching a hit near the beginning of a long
    /// conversation scales with the conversation rather than with the hit.
    ///
    /// Both ends are clamped to the partition's own bounds: `start` up to the
    /// pruning boundary (as [`EventLog::replay_from_with_positions`] does),
    /// `end` down to the journal's tail, so an `end` past the tail reads to
    /// the tail rather than erroring. An empty or inverted range — `start`
    /// at or past the clamped `end` — yields an empty,
    /// `budget_exceeded: false` [`BoundedReplay`], never an error.
    ///
    /// The byte cap keeps the same meaning it has on the sibling methods: the
    /// event that trips the budget is INCLUDED, and `budget_exceeded` is set
    /// so the caller can tell a truncated read from a complete one. A range
    /// that ends before the cap trips returns `budget_exceeded: false` even
    /// if `bytes_read` is large, because the range, not the budget, is what
    /// stopped it.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the journal cannot start the
    /// replay stream or if decoding any stored item fails before either bound
    /// stops it.
    // See [`EventLog::replay_with_positions`]: the snapshot is owned, so the
    // journal lock is released before the stream is consumed.
    pub async fn replay_range_with_positions_bounded(
        &self,
        start: u64,
        end: u64,
        max_bytes: u64,
    ) -> Result<BoundedReplay, EventLogError> {
        let reader = self.journal.lock().await.snapshot().await?;
        let bounds = reader.bounds();
        let from = start.max(bounds.start);
        let until = end.min(bounds.end);
        if from >= until {
            return Ok(BoundedReplay {
                events: Vec::new(),
                bytes_read: 0,
                budget_exceeded: false,
            });
        }
        let stream = reader.replay(from, REPLAY_BUFFER).await?;
        futures::pin_mut!(stream);
        let mut events = Vec::new();
        let mut bytes_read: u64 = 0;
        let mut budget_exceeded = false;
        while let Some(item) = stream.next().await {
            let (position, event) = item?;
            // Checked BEFORE accounting: an event at or past `end` is outside
            // the requested range, so it must not reach the caller and must
            // not spend the caller's byte budget either.
            if position >= until {
                break;
            }
            bytes_read = bytes_read.saturating_add(event.payload.len() as u64);
            events.push((position, event));
            if bytes_read > max_bytes {
                budget_exceeded = true;
                break;
            }
        }
        Ok(BoundedReplay {
            events,
            bytes_read,
            budget_exceeded,
        })
    }

    /// Replay every event in append order, discarding positions.
    ///
    /// Convenience over [`EventLog::replay_with_positions`] for callers that
    /// only need the ordered events.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] on the same conditions as
    /// [`EventLog::replay_with_positions`].
    pub async fn replay(&self) -> Result<Vec<Event>, EventLogError> {
        Ok(self
            .replay_with_positions()
            .await?
            .into_iter()
            .map(|(_pos, event)| event)
            .collect())
    }

    /// Replay events from position `start` in append order, discarding
    /// positions. Convenience over [`EventLog::replay_from_with_positions`].
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] on the same conditions as
    /// [`EventLog::replay_from_with_positions`].
    pub async fn replay_from(&self, start: u64) -> Result<Vec<Event>, EventLogError> {
        Ok(self
            .replay_from_with_positions(start)
            .await?
            .into_iter()
            .map(|(_pos, event)| event)
            .collect())
    }

    /// Replay every event in append order, skipping any position whose item
    /// cannot be decoded rather than aborting the whole replay.
    ///
    /// [`EventLog::replay_with_positions`] stops at the first bad item (the
    /// backup/DR gap #799 tracks: one corrupted event permanently locks a
    /// conversation out of replay). This reads each position independently
    /// through the journal's position index — a corrupted item's neighbors
    /// don't depend on decoding it — so it recovers everything readable and
    /// reports the rest as [`QuarantinedItem`]s. This is the primitive a
    /// `conversation repair` operation uses to drop only the unreadable
    /// event(s) and let the rest of the log replay again; it is otherwise
    /// intended for that recovery path, not routine replay (one read per
    /// position, versus one streamed pass).
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the journal cannot report its
    /// own bounds. A per-item decode failure is reported in the returned
    /// quarantine list, never as an `Err`.
    pub async fn replay_quarantining(
        &self,
    ) -> Result<(Vec<(u64, Event)>, Vec<QuarantinedItem>), EventLogError> {
        let reader = self.journal.lock().await.snapshot().await?;
        let bounds = reader.bounds();
        let mut ok = Vec::new();
        let mut quarantined = Vec::new();
        for position in bounds {
            match reader.read(position).await {
                Ok(event) => ok.push((position, event)),
                Err(err) => quarantined.push(QuarantinedItem {
                    position,
                    error: err.to_string(),
                }),
            }
        }
        Ok((ok, quarantined))
    }

    /// Durably persist all buffered appends, guaranteeing they survive a crash.
    ///
    /// Committed appends survive a crash, but the next [`EventLog::open`] may
    /// perform recovery work rebuilding the position index from data before
    /// replay is available — [`EventLog::sync`] additionally makes the next
    /// open recovery-free.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the underlying flush fails.
    pub async fn commit(&self) -> Result<(), EventLogError> {
        Ok(self.journal.lock().await.commit().await?)
    }

    /// Stronger durability than [`EventLog::commit`]: persist and guarantee no
    /// recovery work is needed on next open.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the underlying sync fails.
    pub async fn sync(&self) -> Result<(), EventLogError> {
        Ok(self.journal.lock().await.sync().await?)
    }
}

#[cfg(test)]
mod tests {
    use super::{Event, EventLog, EventLogConfig, EventLogError};
    use commonware_runtime::{Runner, Supervisor as _, deterministic};

    /// Mirrors the crate-level `# Example` doctest verbatim. The doc block is
    /// marked `ignore` because a runnable doctest links the whole Commonware
    /// stack into its own binary, exhausting CI's linker; this test re-verifies the
    /// same code in the crate's already-linked test binary so the documented
    /// example can't silently rot.
    #[test]
    fn doc_example_open_append_replay() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-1"))
                .await
                .expect("open log");

            log.append(&Event::new("user_msg", b"hello".to_vec()))
                .await
                .unwrap();
            log.append(&Event::new("tool_call", b"\x01\x02".to_vec()))
                .await
                .unwrap();
            log.commit().await.unwrap();

            let events = log.replay().await.unwrap();
            assert_eq!(events.len(), 2);
            assert_eq!(events[0].kind, "user_msg");
        });
    }

    /// Append events across several conversation turns, then assert replay
    /// returns them in append (conversation) order with payload bytes intact.
    #[test]
    fn append_then_replay_preserves_order_and_payload() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-order"))
                .await
                .expect("open");

            // Two turns: turn 0 = user_msg + planner_decision; turn 1 =
            // tool_call + tool_result. Appended in conversation order.
            let appended = vec![
                Event::new("user_msg", b"what is 2+2?".to_vec()),
                Event::new("planner_decision", vec![0xde, 0xad]),
                Event::new("tool_call", vec![0x01, 0x02, 0x03]),
                Event::new("tool_result", vec![0xff, 0x00, 0xff]),
            ];
            for (i, event) in appended.iter().enumerate() {
                let pos = log.append(event).await.expect("append");
                assert_eq!(pos, i as u64, "positions are 0-indexed and contiguous");
            }
            log.commit().await.expect("commit");

            assert_eq!(log.len().await, 4);
            assert!(!log.is_empty().await);

            // Replay yields exactly the appended sequence, in order.
            let replayed = log.replay().await.expect("replay");
            assert_eq!(replayed, appended);

            // Positions are ascending and dense.
            let with_pos = log.replay_with_positions().await.expect("replay+pos");
            let positions: Vec<u64> = with_pos.iter().map(|(p, _)| *p).collect();
            assert_eq!(positions, vec![0, 1, 2, 3]);

            // Payload bytes round-trip verbatim.
            assert_eq!(with_pos[2].1.payload, vec![0x01, 0x02, 0x03]);
        });
    }

    /// Bounded replay: `replay_from(start)` seeks to `start` and yields only the
    /// tail `[start, len)`, never re-reading earlier positions — the position-
    /// indexed primitive a checkpointed replay starts from. `replay_from(0)`
    /// equals a full replay; a `start` at/past the end yields nothing.
    #[test]
    fn replay_from_offset_returns_tail_only() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-from"))
                .await
                .expect("open");
            for i in 0..5u8 {
                log.append(&Event::new(format!("k{i}"), vec![i]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            // A full replay sees all five from position 0.
            assert_eq!(log.replay_with_positions().await.expect("replay").len(), 5);

            // `replay_from(2)` starts at the offset, not 0: positions 2,3,4 only.
            let tail = log
                .replay_from_with_positions(2)
                .await
                .expect("replay_from");
            let positions: Vec<u64> = tail.iter().map(|(p, _)| *p).collect();
            assert_eq!(positions, vec![2, 3, 4]);
            assert_eq!(tail[0].1.kind, "k2");
            assert_eq!(tail.last().expect("non-empty").1.kind, "k4");

            // Starting at or past the end yields nothing.
            assert!(log.replay_from(5).await.expect("from end").is_empty());
            assert!(log.replay_from(99).await.expect("past end").is_empty());

            // `replay_from(0)` is exactly a full replay.
            assert_eq!(
                log.replay_from(0).await.expect("from 0"),
                log.replay().await.expect("replay")
            );
        });
    }

    /// Issue #1541's core early-abort proof: [`EventLog::replay_with_positions_bounded`]
    /// stops pulling from the journal's own replay stream the instant
    /// cumulative payload bytes cross the caller's budget — it does NOT
    /// drain the whole partition first and check afterward. Ten events of
    /// exactly 1,000 bytes each (10,000 bytes total) replayed under a 3,500
    /// byte budget must stop after the FOURTH event (4,000 bytes — the
    /// first cumulative total to exceed 3,500), never reaching the
    /// remaining six. This test fails if a future change reintroduces
    /// full-materialize-then-check: it would return all 10 events instead
    /// of 4.
    #[test]
    fn replay_with_positions_bounded_stops_reading_mid_partition_once_the_budget_trips() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-bounded"))
                .await
                .expect("open");
            for i in 0..10u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 1_000]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let bounded = log
                .replay_with_positions_bounded(3_500)
                .await
                .expect("bounded replay");

            assert!(
                bounded.budget_exceeded,
                "3,500-byte budget over a 10,000-byte partition must trip"
            );
            assert_eq!(
                bounded.events.len(),
                4,
                "replay must stop the instant cumulative bytes (4,000 after the 4th event) \
                 cross the 3,500 budget — reading a 5th event (or draining the whole partition) \
                 means the abort happened too late, or not at all"
            );
            assert_eq!(
                bounded.bytes_read, 4_000,
                "bytes_read must reflect exactly the events actually returned, not the whole \
                 partition's real 10,000 bytes"
            );
            assert!(
                bounded.bytes_read < 10_000,
                "peak materialized bytes must stay bounded well under the partition's real \
                 size — the whole point of stopping mid-stream"
            );

            // The returned prefix is still ordered and intact — an early
            // abort must not corrupt what it DID manage to read.
            let positions: Vec<u64> = bounded.events.iter().map(|(p, _)| *p).collect();
            assert_eq!(positions, vec![0, 1, 2, 3]);
        });
    }

    /// The companion happy path: a partition whose whole payload fits under
    /// the budget replays completely, `budget_exceeded` is `false`, and the
    /// result matches [`EventLog::replay_with_positions`] exactly — the byte
    /// budget must never truncate a scope that is genuinely within it.
    #[test]
    fn replay_with_positions_bounded_reads_everything_when_under_budget() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-under-budget"))
                .await
                .expect("open");
            for i in 0..5u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 100]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let unbounded = log.replay_with_positions().await.expect("replay");
            let bounded = log
                .replay_with_positions_bounded(10_000)
                .await
                .expect("bounded replay");

            assert!(
                !bounded.budget_exceeded,
                "500 bytes under a 10,000 byte budget must never trip"
            );
            assert_eq!(
                bounded.events, unbounded,
                "must match the unbounded replay exactly"
            );
            assert_eq!(bounded.bytes_read, 500);
        });
    }

    /// [`EventLog::replay_from_with_positions_bounded`]'s own core proof: it
    /// honors BOTH `start` (skip everything before the caller's watermark,
    /// like [`EventLog::replay_from_with_positions`]) AND `max_bytes` (stop
    /// mid-stream once the budget trips, like
    /// [`EventLog::replay_with_positions_bounded`]) in the same call — the
    /// combined primitive neither of those two alone can express. Ten
    /// 1,000-byte events; resuming at position 3 (skipping the first three,
    /// 3,000 bytes) under a 2,500-byte budget must stop after the SECOND
    /// event it actually reads (positions 3 and 4 — 2,000 bytes), never
    /// reaching position 5, and never re-reading positions 0..3 at all.
    #[test]
    fn replay_from_with_positions_bounded_honors_both_start_and_the_byte_budget() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-tail-bounded"))
                .await
                .expect("open");
            for i in 0..10u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 1_000]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let bounded = log
                .replay_from_with_positions_bounded(3, 2_500)
                .await
                .expect("bounded tail replay");

            assert!(
                bounded.budget_exceeded,
                "a 2,500-byte budget over a 7,000-byte tail (positions 3..10) must trip"
            );
            let positions: Vec<u64> = bounded.events.iter().map(|(p, _)| *p).collect();
            assert_eq!(
                positions,
                vec![3, 4, 5],
                "must resume at position 3 (never re-reading 0..3) and stop on the event that \
                 CROSSES the 2,500 budget — that event is returned, matching \
                 `EventLog::replay_with_positions_bounded`'s own accumulate-push-then-check \
                 order, so the two differ only in where they start"
            );
            assert_eq!(
                bounded.bytes_read, 3_000,
                "bytes_read counts exactly the events actually returned, the budget-crossing \
                 one included — never the tail's whole 7,000 bytes"
            );
        });
    }

    /// `end` is EXCLUSIVE, and the boundary is exact: a range ending at `n`
    /// returns position `n - 1` and never `n`.
    ///
    /// The off-by-one this pins is the whole point of the primitive. A caller
    /// fetching one turn locates `[turn_start, turn_end)` from an index and
    /// must get that span and nothing adjacent — an inclusive end would leak
    /// the first event of the NEXT turn into every fetch.
    #[test]
    fn replay_range_excludes_the_end_position_exactly() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-range-exact"))
                .await
                .expect("open");
            for i in 0..10u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 100]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let bounded = log
                .replay_range_with_positions_bounded(3, 6, u64::MAX)
                .await
                .expect("range replay");

            let positions: Vec<u64> = bounded.events.iter().map(|(p, _)| *p).collect();
            assert_eq!(
                positions,
                vec![3, 4, 5],
                "[3, 6) is positions 3, 4, 5 — position 6 is outside the range and must not be \
                 returned"
            );
            assert!(
                !bounded.budget_exceeded,
                "the RANGE stopped this replay, not the budget; conflating the two would tell a \
                 caller its result was truncated when it is complete"
            );
            assert_eq!(
                bounded.bytes_read, 300,
                "the excluded end event must not spend the caller's byte budget either"
            );
        });
    }

    /// An `end` past the journal's tail clamps to the tail rather than
    /// erroring — the upper-bound mirror of `start`'s own clamp up to the
    /// pruning boundary. A caller holding a stale end position gets what
    /// exists, not a failure.
    #[test]
    fn replay_range_end_past_the_tail_clamps_to_the_tail() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-range-clamp"))
                .await
                .expect("open");
            for i in 0..4u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 100]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let bounded = log
                .replay_range_with_positions_bounded(2, 9_999, u64::MAX)
                .await
                .expect("range replay");

            let positions: Vec<u64> = bounded.events.iter().map(|(p, _)| *p).collect();
            assert_eq!(positions, vec![2, 3], "clamped to the tail, not an error");
            assert!(!bounded.budget_exceeded);
        });
    }

    /// The LOWER-bound mirror of `replay_range_end_past_the_tail_clamps_to_the_tail`
    /// just above: a `start` BELOW the partition's own pruning boundary
    /// clamps UP to it (`start.max(bounds.start)`), rather than asking the
    /// journal to replay from an already-pruned position. The journal
    /// refuses that outright
    /// (`commonware_storage::journal::contiguous::variable`'s
    /// `Error::ItemPruned`, surfacing here as `EventLogError::Journal`), so
    /// without the clamp this call would return `Err`, not the `Ok` this
    /// test asserts.
    ///
    /// `items_per_section` is overridden to 1 for the same reason
    /// `repair_replay_quarantining_recovers_events_after_a_corrupted_earlier_section`
    /// overrides it: the default 1,024-item section makes the pruning
    /// boundary land on a multiple of 1,024, far past anything a fast unit
    /// test could plausibly append — with one item per section, pruning to
    /// position 3 lands the boundary exactly on 3.
    #[test]
    fn replay_range_start_below_the_pruning_boundary_clamps_up_to_it() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut cfg = EventLogConfig::for_partition("conv-range-pruned");
            cfg.items_per_section = commonware_utils::NZU64!(1);
            let log = EventLog::open(context, cfg).await.expect("open");
            for i in 0..6u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 100]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            // Prune positions 0..3 — the store's own pruning boundary, not a
            // caller-held watermark. Accessed via the underlying journal
            // directly: `EventLog` has no public prune wrapper of its own,
            // and this test module is the crate root's own child, so the
            // private `journal` field is reachable here.
            let pruned = log.journal.lock().await.prune(3).await.expect("prune");
            assert!(
                pruned,
                "positions 0..3 must actually have been pruned — otherwise the clamp below is \
                 never exercised"
            );

            let bounded = log
                .replay_range_with_positions_bounded(1, 5, u64::MAX)
                .await
                .expect("a start below the pruning boundary must clamp up to it, never error");

            let positions: Vec<u64> = bounded.events.iter().map(|(p, _)| *p).collect();
            assert_eq!(
                positions,
                vec![3, 4],
                "clamped up to the pruning boundary (3), not the caller's stale start (1) — and \
                 still respecting the requested end (5)"
            );
            assert!(!bounded.budget_exceeded);
        });
    }

    /// An empty or inverted range yields an empty, non-exceeded
    /// [`BoundedReplay`] — never an error. Equal bounds are empty because the
    /// end is exclusive; an inverted range is empty rather than being
    /// silently reordered into a real one.
    #[test]
    fn replay_range_empty_and_inverted_are_empty_not_errors() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-range-empty"))
                .await
                .expect("open");
            for i in 0..5u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 100]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            for (start, end, why) in [
                (2u64, 2u64, "equal bounds are empty — the end is exclusive"),
                (4, 1, "an inverted range is empty, never reordered"),
                (99, 200, "a range entirely past the tail is empty"),
            ] {
                let bounded = log
                    .replay_range_with_positions_bounded(start, end, u64::MAX)
                    .await
                    .expect("range replay");
                assert!(bounded.events.is_empty(), "{why}");
                assert_eq!(bounded.bytes_read, 0, "{why}");
                assert!(
                    !bounded.budget_exceeded,
                    "an empty range must not report a tripped budget: {why}"
                );
            }
        });
    }

    /// The byte cap still applies WITHIN a range, and stops the replay before
    /// the range does — so a caller can tell "the range ended" from "I ran out
    /// of budget" by `budget_exceeded` alone.
    #[test]
    fn replay_range_byte_cap_trips_inside_the_range() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-range-budget"))
                .await
                .expect("open");
            for i in 0..10u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 1_000]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let bounded = log
                .replay_range_with_positions_bounded(2, 9, 2_500)
                .await
                .expect("range replay");

            assert!(
                bounded.budget_exceeded,
                "a 2,500-byte budget over a 7,000-byte range must trip"
            );
            let positions: Vec<u64> = bounded.events.iter().map(|(p, _)| *p).collect();
            assert_eq!(
                positions,
                vec![2, 3, 4],
                "stops on the event that CROSSES the budget, which is returned — the same \
                 accumulate-push-then-check order the sibling bounded replays use, so the three \
                 differ only in where they start and stop"
            );
            assert_eq!(bounded.bytes_read, 3_000);
        });
    }

    /// The empty-range case: a `start` at or past the partition's end yields
    /// an empty, non-exceeded [`BoundedReplay`] — never an error, and never a
    /// spurious `budget_exceeded` — mirroring
    /// [`EventLog::replay_from_with_positions`]'s own empty-range rule.
    #[test]
    fn replay_from_with_positions_bounded_past_the_end_is_empty_not_exceeded() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(
                context,
                EventLogConfig::for_partition("conv-tail-bounded-empty"),
            )
            .await
            .expect("open");
            for i in 0..3u32 {
                log.append(&Event::new(format!("k{i}"), vec![0u8; 1_000]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let bounded = log
                .replay_from_with_positions_bounded(99, 1)
                .await
                .expect("bounded tail replay past the end");

            assert!(bounded.events.is_empty());
            assert_eq!(bounded.bytes_read, 0);
            assert!(
                !bounded.budget_exceeded,
                "an empty replay must never report the budget as exceeded"
            );
        });
    }

    /// A healthy log's quarantining replay reports every event decoded and
    /// nothing quarantined — the repair primitive's happy path.
    #[test]
    fn repair_replay_quarantining_reports_nothing_bad_on_a_healthy_log() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-healthy"))
                .await
                .expect("open");
            for i in 0..4u8 {
                log.append(&Event::new(format!("k{i}"), vec![i]))
                    .await
                    .expect("append");
            }
            log.commit().await.expect("commit");

            let (ok, quarantined) = log.replay_quarantining().await.expect("quarantine replay");
            assert_eq!(ok.len(), 4);
            assert!(quarantined.is_empty());
            assert_eq!(ok, log.replay_with_positions().await.expect("replay"));
        });
    }

    /// A freshly opened log is empty and replays nothing.
    #[test]
    fn empty_log_replays_empty() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let log = EventLog::open(context, EventLogConfig::for_partition("conv-empty"))
                .await
                .expect("open");
            assert!(log.is_empty().await);
            assert_eq!(log.len().await, 0);
            assert!(log.replay().await.expect("replay").is_empty());
        });
    }

    /// Reset empties the partition and hands back a journal the caller can
    /// keep using: position zero, no surviving events, and the emptiness
    /// outlives the handle.
    ///
    /// The reuse half of the erasure primitive (`#216`) — a rewrite, repair,
    /// or migration empties a partition and then refills it, so the reset has
    /// to return something to refill.
    #[test]
    fn reset_empties_the_partition_and_returns_a_usable_journal() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = EventLogConfig::for_partition("reset-me");
            let log = EventLog::open(context.child("first"), cfg.clone())
                .await
                .expect("open");
            log.append(&Event::new("k", b"payload".to_vec()))
                .await
                .expect("append");
            log.commit().await.expect("commit");

            let log = log.reset(context.child("first")).await.expect("reset");
            assert_eq!(log.len().await, 0, "the reset journal is empty");
            assert!(log.replay().await.expect("replay").is_empty());

            let pos = log
                .append(&Event::new("k2", b"fresh".to_vec()))
                .await
                .expect("append after reset");
            assert_eq!(pos, 0, "the reset journal restarts at position zero");
            log.commit().await.expect("commit");
            drop(log);

            let log = EventLog::open(context.child("second"), cfg)
                .await
                .expect("reopen");
            let replayed = log.replay().await.expect("replay");
            assert_eq!(
                replayed.len(),
                1,
                "the reset is durable: only the post-reset event survives"
            );
            assert_eq!(replayed[0].kind, "k2");
        });
    }

    /// Reset then reclaim removes the partition wholesale: a reopen starts
    /// empty, and appends after the reopen work normally (no resurrection of
    /// old events). The erasure half of `#216`.
    #[test]
    fn reclaim_removes_the_partition_and_reopen_is_empty() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = EventLogConfig::for_partition("reclaim-me");
            let log = EventLog::open(context.child("first"), cfg.clone())
                .await
                .expect("open");
            log.append(&Event::new("k", b"payload".to_vec()))
                .await
                .expect("append");
            log.commit().await.expect("commit");

            let log = log.reset(context.child("first")).await.expect("reset");
            log.reclaim().await.expect("reclaim");

            let log = EventLog::open(context.child("second"), cfg)
                .await
                .expect("reopen");
            assert!(
                log.replay().await.expect("replay").is_empty(),
                "a reclaimed partition must reopen empty"
            );
            let pos = log
                .append(&Event::new("k2", b"fresh".to_vec()))
                .await
                .expect("append after reclaim");
            assert_eq!(pos, 0, "the fresh partition starts at position zero");
        });
    }

    /// Reclaim refuses a journal that still holds events.
    ///
    /// Reclaim is Commonware's final-teardown removal, which is not crash
    /// safe. The refusal is what keeps it off content: a caller that skipped
    /// the reset gets an error, not an interrupted removal that could leave a
    /// short readable history behind.
    #[test]
    fn reclaim_refuses_a_journal_that_still_holds_events() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = EventLogConfig::for_partition("reclaim-refuses");
            let log = EventLog::open(context.child("first"), cfg.clone())
                .await
                .expect("open");
            log.append(&Event::new("k", b"payload".to_vec()))
                .await
                .expect("append");
            log.commit().await.expect("commit");

            let error = log.reclaim().await.expect_err("reclaim must refuse");
            assert!(
                matches!(error, EventLogError::ReclaimNotEmpty { events: 1 }),
                "expected a refusal naming the surviving event count, got {error}"
            );

            let log = EventLog::open(context.child("second"), cfg)
                .await
                .expect("reopen");
            assert_eq!(
                log.replay().await.expect("replay").len(),
                1,
                "the refused reclaim removed nothing"
            );
        });
    }

    /// Pins the `sync()` durability contract: after a `sync`, reopen needs no
    /// recovery work and returns exactly the synced events. (`commit()`
    /// alone is a distinct, weaker contract — see
    /// `reopen_recovers_committed_events_after_commit_only` below, which
    /// pins that one instead.)
    #[test]
    fn reopen_recovers_synced_events() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = EventLogConfig::for_partition("conv-reopen");

            {
                // Distinct supervision-tree label per open simulates a separate
                // process (the deterministic runtime's metric registry is
                // shared for the whole run, so re-registering under the same
                // label panics — a real restart gets a fresh registry).
                let log = EventLog::open(context.child("first"), cfg.clone())
                    .await
                    .expect("open first");
                log.append(&Event::new("user_msg", b"persist me".to_vec()))
                    .await
                    .expect("append");
                log.sync().await.expect("sync");
            } // drop the handle

            let log = EventLog::open(context.child("second"), cfg)
                .await
                .expect("reopen");
            let replayed = log.replay().await.expect("replay");
            assert_eq!(replayed.len(), 1);
            assert_eq!(replayed[0].kind, "user_msg");
            assert_eq!(replayed[0].payload, b"persist me".to_vec());
        });
    }

    /// Pins the explicit 2026.7 `commit()`-only durability contract: every
    /// production batch boundary (`append_batch_one` in `polyc-eventlog-host`)
    /// ends in `commit()`, never `sync()`. Upstream's `commit()` fsyncs dirty data blobs but does not
    /// advance the offsets recovery watermark, so reopen after a commit-only
    /// crash may perform recovery work — rebuilding the missing offset
    /// entries by replaying data from the recovery anchor — rather than
    /// finding a synced index ready to go. This test crashes deliberately
    /// between `commit()` and any `sync()` (the handle is simply dropped, on
    /// a real OS-backed runtime so the process boundary is genuine, not
    /// simulated in-memory state) and asserts the committed events are still
    /// fully recovered on reopen, in order, with appends able to continue
    /// past them. This catches a regression where committed bytes never
    /// leave the application-level tail buffer at all (an in-process reopen
    /// can't tell a real `fsync` apart from a plain unsynced `write()`, since
    /// the OS page cache survives process death, not just power loss) or
    /// where the offsets-rebuild recovery path breaks; the fsync guarantee
    /// itself rests on reading upstream's `commit()` source, not on this
    /// test.
    #[test]
    fn reopen_recovers_committed_events_after_commit_only() {
        use commonware_runtime::{Runner as _, tokio as cw_tokio};

        let dir =
            std::env::temp_dir().join(format!("polyc-eventlog-commit-only-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let cfg = EventLogConfig::for_partition("conv-commit-only");

        let write_cfg = cfg.clone();
        let write_runner =
            cw_tokio::Runner::new(cw_tokio::Config::default().with_storage_directory(dir.clone()));
        write_runner.start(move |context| async move {
            let log = EventLog::open(context, write_cfg).await.expect("open");
            log.append(&Event::new("user_msg", b"one".to_vec()))
                .await
                .expect("append 0");
            log.append(&Event::new("output_msg", b"two".to_vec()))
                .await
                .expect("append 1");
            log.append(&Event::new("tool_call", b"three".to_vec()))
                .await
                .expect("append 2");
            log.commit().await.expect("commit");
            // No `sync()` — the runner ends here and the handle drops,
            // simulating a crash immediately after the commit-only durability
            // point every production batch boundary actually uses.
        });

        let read_cfg = cfg;
        let read_runner =
            cw_tokio::Runner::new(cw_tokio::Config::default().with_storage_directory(dir.clone()));
        read_runner.start(move |context| async move {
            let log = EventLog::open(context, read_cfg)
                .await
                .expect("reopen recovers committed-only data with no sync");
            let replayed = log.replay().await.expect("replay");
            assert_eq!(replayed.len(), 3, "all three committed events survive");
            assert_eq!(replayed[0].kind, "user_msg");
            assert_eq!(replayed[0].payload, b"one".to_vec());
            assert_eq!(replayed[1].kind, "output_msg");
            assert_eq!(replayed[1].payload, b"two".to_vec());
            assert_eq!(replayed[2].kind, "tool_call");
            assert_eq!(replayed[2].payload, b"three".to_vec());

            let pos = log
                .append(&Event::new("recovered", b"still writable".to_vec()))
                .await
                .expect("append continues after commit-only recovery");
            assert_eq!(pos, 3, "the new append continues at position 3");
        });

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Deterministic-runtime sibling of
    /// `reopen_recovers_committed_events_after_commit_only`: the crate
    /// documents both the deterministic and tokio runtimes, so the
    /// commit-only durability contract is pinned on both. Cheaper than the
    /// tokio version (no real filesystem I/O) but does not exercise a real
    /// OS-backed crash boundary — that's what the tokio sibling is for.
    #[test]
    fn reopen_recovers_committed_events_after_commit_only_deterministic() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = EventLogConfig::for_partition("conv-reopen-commit-only");

            {
                let log = EventLog::open(context.child("first"), cfg.clone())
                    .await
                    .expect("open first");
                log.append(&Event::new("user_msg", b"persist me".to_vec()))
                    .await
                    .expect("append");
                log.commit().await.expect("commit");
                // No `sync()` before drop.
            }

            let log = EventLog::open(context.child("second"), cfg)
                .await
                .expect("reopen");
            let replayed = log.replay().await.expect("replay");
            assert_eq!(replayed.len(), 1);
            assert_eq!(replayed[0].kind, "user_msg");
            assert_eq!(replayed[0].payload, b"persist me".to_vec());
        });
    }

    /// Determinism / reproducibility: two independent deterministic runs with
    /// the same seeded program produce the same auditor state. This is the
    /// property replay tests rely on (mirrors the runtime spike's
    /// `auditor().state()` assertion).
    #[test]
    fn deterministic_runs_are_reproducible() {
        fn run() -> String {
            let executor = deterministic::Runner::default();
            executor.start(|context| async move {
                // `Context` is no longer `Clone` in 2026.5 — use `child`
                // to produce a sibling context for the log while keeping
                // the parent available for the `auditor()` read at the end.
                let log =
                    EventLog::open(context.child("det"), EventLogConfig::for_partition("det"))
                        .await
                        .expect("open");
                for i in 0..6u8 {
                    log.append(&Event::new(format!("kind-{i}"), vec![i; i as usize]))
                        .await
                        .expect("append");
                }
                log.commit().await.expect("commit");
                let _ = log.replay().await.expect("replay");
                context.auditor().state()
            })
        }

        let first = run();
        let second = run();
        assert_eq!(first, second, "deterministic runtime must be reproducible");
    }

    /// The repair primitive (`#799`), on real on-disk files: `EventLog::open`
    /// only re-validates its FINAL (still-active) section on open — see the
    /// crate docs — so a corrupted item in an EARLIER, already-closed
    /// section is invisible to that recovery and `open` succeeds anyway. A
    /// plain streaming [`EventLog::replay`] then HARD FAILS as soon as it
    /// streams past the corrupted item (commonware 2026.7 tightened this:
    /// 2026.5 silently dropped the corrupted item with no error signal at
    /// all — an even worse gap, since a caller saw a shorter-than-expected
    /// transcript with no indication anything was missing). Because the
    /// corrupted position here is the earliest one, the whole replay
    /// aborts and the later, undamaged events (1 and 2) are inaccessible
    /// through this path too. [`EventLog::replay_quarantining`] reads each
    /// position independently through the journal's offset index,
    /// correctly reports the corrupted position (with the underlying
    /// storage error), and still recovers everything after it — the
    /// primitive this repair path exists for.
    #[test]
    fn repair_replay_quarantining_recovers_events_after_a_corrupted_earlier_section() {
        use commonware_runtime::{Runner as _, tokio as cw_tokio};

        let dir = std::env::temp_dir().join(format!(
            "polyc-eventlog-repair-earlier-section-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);

        // One event per section, so item 0 lands in a section that is no
        // longer "final" (and so no longer re-validated) once items 1 and 2
        // are appended after it.
        let mut cfg = EventLogConfig::for_partition("conv-repair");
        cfg.items_per_section = commonware_utils::NZU64!(1);

        let write_cfg = cfg.clone();
        let write_runner =
            cw_tokio::Runner::new(cw_tokio::Config::default().with_storage_directory(dir.clone()));
        write_runner.start(move |context| async move {
            let log = EventLog::open(context, write_cfg).await.expect("open");
            log.append(&Event::new("user_msg", vec![b'A'; 20]))
                .await
                .expect("append 0");
            log.append(&Event::new("output_msg", vec![b'B'; 20]))
                .await
                .expect("append 1");
            log.append(&Event::new("tool_call", vec![b'C'; 20]))
                .await
                .expect("append 2");
            log.sync().await.expect("sync");
        });

        // Corrupt section 0's item at the first byte of its `kind` STRING
        // content (byte offset 10: 8-byte section magic header + 1-byte
        // outer item-length prefix + 1-byte inner kind-length prefix),
        // leaving both length prefixes intact so the outer framing still
        // parses fine. `0xFF` is not a valid UTF-8 lead byte in any
        // position, and [`Event`]'s decode is strict (`String::from_utf8`,
        // not lossy), so this is a genuine decode failure — simulating bit
        // rot or tampering in an already-closed section — rather than
        // silently decoding into different-but-still-valid content (the
        // kind of tamper only the MMR check in `polyc_eventlog::integrity`
        // catches, not this decode-level quarantine).
        let data_file = dir.join("conv-repair_data").join("0000000000000000");
        let mut bytes = std::fs::read(&data_file).expect("read section 0");
        bytes[10] = 0xFF;
        std::fs::write(&data_file, &bytes).expect("write corrupted section 0");

        let read_cfg = cfg;
        let read_runner =
            cw_tokio::Runner::new(cw_tokio::Config::default().with_storage_directory(dir.clone()));
        read_runner.start(move |context| async move {
            let log = EventLog::open(context, read_cfg)
                .await
                .expect("reopen succeeds: only the final section is re-validated on open");

            let plain_err = log
                .replay()
                .await
                .expect_err("a corrupted item makes the whole streamed replay fail");
            assert!(
                matches!(plain_err, EventLogError::Journal(_)),
                "unexpected error variant: {plain_err:?}"
            );

            let (ok, quarantined) = log
                .replay_quarantining()
                .await
                .expect("quarantining replay reports positions, not an Err");
            assert_eq!(quarantined.len(), 1, "exactly the corrupted item");
            assert_eq!(quarantined[0].position, 0);
            assert_eq!(
                ok.len(),
                2,
                "the two events after the corrupted one recover"
            );
            assert_eq!(ok[0], (1, Event::new("output_msg", vec![b'B'; 20])));
            assert_eq!(ok[1], (2, Event::new("tool_call", vec![b'C'; 20])));
        });

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Torn-write recovery (`#799`): a crash mid-append leaves a partition
    /// whose on-disk section no longer matches what the offset index
    /// expects. Reopening must not error or panic, replay must return
    /// successfully (never hard-fail the whole partition open over a crash
    /// artifact), and the partition must accept new appends afterward — the
    /// durability property every append-only log needs to survive a real
    /// power loss / OOM-kill: a torn write degrades the log, it does not
    /// permanently brick the conversation.
    #[test]
    fn torn_write_truncated_journal_reopens_and_replays() {
        use commonware_runtime::{Runner as _, tokio as cw_tokio};

        let dir =
            std::env::temp_dir().join(format!("polyc-eventlog-torn-write-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let cfg = EventLogConfig::for_partition("conv-torn");

        let write_cfg = cfg.clone();
        let write_runner =
            cw_tokio::Runner::new(cw_tokio::Config::default().with_storage_directory(dir.clone()));
        write_runner.start(move |context| async move {
            let log = EventLog::open(context, write_cfg).await.expect("open");
            log.append(&Event::new("user_msg", vec![b'A'; 20]))
                .await
                .expect("append 0");
            log.append(&Event::new("output_msg", vec![b'B'; 20]))
                .await
                .expect("append 1");
            log.append(&Event::new("tool_call", vec![b'C'; 20]))
                .await
                .expect("append 2");
            log.sync().await.expect("sync");
        });

        // Simulate a crash mid-write: the blob's storage space is
        // pre-allocated well beyond the ~107 bytes the three items occupy,
        // so a real crash leaves the file at its full pre-allocated length
        // with an un-written (zero) tail rather than a shorter file. Locate
        // item 2's payload by content (rather than hand-deriving on-disk
        // item-framing offsets) and zero everything from partway through it
        // onward — a torn (incomplete) final item, exactly what a crash
        // mid-append-of-item-2 leaves behind.
        let data_file = dir.join("conv-torn_data").join("0000000000000000");
        let full = std::fs::read(&data_file).expect("read section 0");
        let marker = [b'C'; 20];
        let payload_start = full
            .windows(marker.len())
            .position(|w| w == marker)
            .expect("item 2's payload is present in the untruncated section");
        let mut corrupted = full;
        for b in &mut corrupted[payload_start + 10..] {
            *b = 0;
        }
        std::fs::write(&data_file, &corrupted).expect("simulate a torn write");

        let read_cfg = cfg;
        let read_runner =
            cw_tokio::Runner::new(cw_tokio::Config::default().with_storage_directory(dir.clone()));
        read_runner.start(move |context| async move {
            let log = EventLog::open(context, read_cfg)
                .await
                .expect("reopen recovers from a torn tail without erroring");
            let events = log
                .replay()
                .await
                .expect("replay succeeds (never hard-fails) after a torn write");
            // The engine's own crash-recovery decides how much of the torn
            // section it can trust; this pins the property that matters for
            // durability — recovery is conservative (it never returns
            // content associated with an item it couldn't fully validate),
            // and it never returns more than what was actually written.
            assert!(
                events.len() <= 3,
                "recovery must never fabricate events beyond what was appended"
            );
            for event in &events {
                assert!(
                    [
                        Event::new("user_msg", vec![b'A'; 20]),
                        Event::new("output_msg", vec![b'B'; 20]),
                    ]
                    .contains(event),
                    "recovery must never return the torn (never-fully-written) third item"
                );
            }

            // The partition is not permanently bricked: it still accepts
            // new appends after the crash.
            let pos = log
                .append(&Event::new("recovered", b"still writable".to_vec()))
                .await
                .expect("the partition accepts appends again after torn-write recovery");
            log.commit().await.expect("commit after recovery");
            assert_eq!(
                pos,
                events.len() as u64,
                "the new append continues from wherever recovery left off"
            );
        });

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Every backing blob file under one of `partition`'s storage
    /// directories, sorted so a failure names the same file twice.
    fn backing_blobs(
        dir: &std::path::Path,
        suffix: &str,
        partition: &str,
    ) -> Vec<std::path::PathBuf> {
        let mut found: Vec<std::path::PathBuf> =
            std::fs::read_dir(dir.join(format!("{partition}{suffix}")))
                .expect("partition storage directory")
                .map(|entry| entry.expect("directory entry").path())
                .filter(|path| path.is_file())
                .collect();
        found.sort();
        found
    }

    /// Seed one committed partition of three events under `dir`.
    fn seed_three_events(dir: &std::path::Path, cfg: &EventLogConfig) {
        use commonware_runtime::{Runner as _, tokio as cw_tokio};

        let cfg = cfg.clone();
        let runner = cw_tokio::Runner::new(
            cw_tokio::Config::default().with_storage_directory(dir.to_path_buf()),
        );
        runner.start(move |context| async move {
            let log = EventLog::open(context, cfg).await.expect("open");
            for i in 0..3u32 {
                log.append(&Event::new(format!("k{i}"), vec![b'x'; 32]))
                    .await
                    .expect("append");
            }
            log.sync().await.expect("sync");
        });
    }

    /// How a reopen answered after a backing blob lost its runtime header.
    #[derive(Debug, PartialEq, Eq)]
    enum Reopened {
        /// The reopen failed outright.
        Refused,
        /// The reopen succeeded and the log holds this many events.
        Holding(usize),
    }

    /// Reopen `partition` under `dir` and report what it holds.
    fn reopen_and_count(dir: &std::path::Path, cfg: &EventLogConfig) -> Reopened {
        use commonware_runtime::{Runner as _, tokio as cw_tokio};

        let cfg = cfg.clone();
        let runner = cw_tokio::Runner::new(
            cw_tokio::Config::default().with_storage_directory(dir.to_path_buf()),
        );
        runner.start(move |context| async move {
            let Ok(log) = EventLog::open(context, cfg).await else {
                return Reopened::Refused;
            };
            log.replay()
                .await
                .map_or(Reopened::Refused, |events| Reopened::Holding(events.len()))
        })
    }

    /// Commonware 2026.7.1 turns every backing blob shorter than its
    /// eight-byte runtime header into a valid EMPTY blob: `Header::missing`
    /// is true below eight bytes, and the tokio backend then truncates the
    /// file to zero, writes a fresh header, and syncs it
    /// (`runtime/src/storage/tokio/mod.rs::Storage::open_versioned`). That is
    /// deliberate recovery of an interrupted first creation.
    ///
    /// The primitive has no way to tell an interrupted creation from an
    /// ESTABLISHED blob later shortened to zero through seven bytes. This
    /// test pins what the primitive actually does to an established
    /// partition, for both of its backing blobs and for every prefix length,
    /// on the real filesystem backend the release changed.
    ///
    /// The measured answer is not the same for the two blobs, and the
    /// difference is the whole reason the layers above cannot delegate this:
    ///
    /// - the OFFSETS blob shortened to any of the eight lengths is REFUSED.
    ///   Commonware will not mount a journal whose offsets it cannot
    ///   reconcile, so that half fails closed on its own.
    /// - the DATA blob shortened to any of the eight lengths reopens as a
    ///   healthy, EMPTY journal, with no error. An established three-event
    ///   conversation becomes an empty one, and nothing at this layer can
    ///   tell it from a conversation that never held anything.
    ///
    /// That second row is not a defect in Commonware — it is documented
    /// recovery of an interrupted creation, and the primitive has no evidence
    /// to separate the two cases. It IS the reason this crate is not the
    /// layer that decides absence. The durable event-count floor is, and
    /// `polyc-eventlog-host` and the state plane carry the tests that prove
    /// the floor turns this empty answer back into damage.
    ///
    /// The invariant asserted alongside the exact rows is that no length
    /// produces a SHORT log. A partial history is the one answer no layer
    /// above could tell from a legitimately short conversation.
    #[test]
    fn a_backing_blob_below_the_runtime_header_never_reopens_short() {
        let root = std::env::temp_dir().join(format!(
            "polyc-eventlog-partial-header-{}-{}",
            std::process::id(),
            "established"
        ));
        let _ = std::fs::remove_dir_all(&root);

        let mut observed = Vec::new();
        for suffix in ["_data", "_offsets-blobs"] {
            for raw_len in 0..8u64 {
                let dir = root.join(format!("{}{raw_len}", suffix.trim_start_matches('_')));
                std::fs::create_dir_all(&dir).expect("case directory");
                let cfg = EventLogConfig::for_partition("conv-partial-header");
                seed_three_events(&dir, &cfg);

                let blobs = backing_blobs(&dir, suffix, "conv-partial-header");
                assert!(
                    !blobs.is_empty(),
                    "a committed partition must leave a {suffix} blob to shorten"
                );
                for blob in &blobs {
                    let file = std::fs::OpenOptions::new()
                        .write(true)
                        .open(blob)
                        .expect("open the backing blob");
                    file.set_len(raw_len).expect("shorten below the header");
                    file.sync_all().expect("make the truncation durable");
                }

                let outcome = reopen_and_count(&dir, &cfg);
                assert!(
                    matches!(outcome, Reopened::Refused | Reopened::Holding(0 | 3)),
                    "{suffix} shortened to {raw_len} byte(s) reopened SHORT as {outcome:?}; a \
                     partial history is the one answer no layer above can tell from a \
                     legitimately short conversation"
                );
                let expected = match suffix {
                    "_data" => Reopened::Holding(0),
                    _ => Reopened::Refused,
                };
                assert_eq!(
                    outcome, expected,
                    "{suffix} shortened to {raw_len} byte(s) answered {outcome:?}. This test \
                     pins what Commonware 2026.7.1 actually does, so a release that changes it \
                     fails here rather than silently moving what the layers above have to catch"
                );
                observed.push((suffix, raw_len, outcome));
                let _ = std::fs::remove_dir_all(&dir);
            }
        }

        let _ = std::fs::remove_dir_all(&root);
        assert_eq!(
            observed.len(),
            16,
            "eight prefix lengths for each of two blobs"
        );
    }

    /// An interrupted reset never leaves a partial history, and the storage it
    /// leaves behind is finished by the next open — across a process restart,
    /// under a real storage fault, for every operation the reset performs.
    ///
    /// This is the crash-safety claim `reset` exists to make. Commonware's
    /// `Mutable::destroy` cannot make it: its own contract says an interrupted
    /// destroy "may observe partially removed state". `init_at_size` stages
    /// the offsets reset intent durably BEFORE clearing data, so every
    /// interruption leaves either the old history intact (the reset never
    /// started) or an empty journal (the reset is finished by the next
    /// `init`). Never one, never two.
    ///
    /// The two runs are two processes: the fault is armed and the reset
    /// attempted in the first, and the assertion is made in the second, over
    /// the storage the first left behind. `start_and_recover` carries that
    /// storage across, which is the recovery boundary
    /// `commonware_runtime`'s own testing guide names.
    ///
    /// The faults are real. Each rate makes the deterministic runtime's
    /// storage fail that operation outright, so the reset fails inside
    /// Commonware rather than at a seam that returned before it wrote
    /// anything.
    #[test]
    fn an_interrupted_reset_survives_a_process_restart_without_a_partial_history() {
        use commonware_runtime::{Runner as _, deterministic, deterministic::FaultConfig};

        let cases = [
            ("open", FaultConfig::default().open(1.0)),
            ("write", FaultConfig::default().write(1.0)),
            ("sync", FaultConfig::default().sync(1.0)),
            ("resize", FaultConfig::default().resize(1.0)),
            ("remove", FaultConfig::default().remove(1.0)),
            ("scan", FaultConfig::default().scan(1.0)),
            (
                "torn write",
                FaultConfig::default().write(1.0).partial_write(1.0),
            ),
            (
                "torn resize",
                FaultConfig::default().resize(1.0).partial_resize(1.0),
            ),
        ];

        let mut outcomes = Vec::new();
        for (label, faults) in cases {
            let partition = format!("conv-reset-crash-{}", label.replace(' ', "-"));

            // Run one: seed a committed history, then attempt a reset with the
            // fault armed. Whether the reset reports success or failure is not
            // the claim; what it leaves on disk is.
            let seeded = partition.clone();
            let runner = deterministic::Runner::timed(std::time::Duration::from_secs(30));
            let ((), checkpoint) = runner.start_and_recover(move |context| async move {
                let cfg = EventLogConfig::for_partition(seeded);
                let log = EventLog::open(context.child("seed"), cfg.clone())
                    .await
                    .expect("open");
                for i in 0..3u32 {
                    log.append(&Event::new(format!("k{i}"), vec![b'x'; 24]))
                        .await
                        .expect("append");
                }
                log.sync().await.expect("sync");
                drop(log);

                let log = EventLog::open(context.child("reset"), cfg)
                    .await
                    .expect("reopen before the reset");
                *context.storage_fault_config().write() = faults;
                // Consuming either way: on failure the handle is gone with the
                // error, which is what Commonware's rule for a failed mutable
                // operation requires.
                drop(log.reset(context.child("reset")).await.ok());
                *context.storage_fault_config().write() = FaultConfig::default();
            });

            // Run two: a different process over the storage run one left.
            let restarted = partition.clone();
            let runner = deterministic::Runner::from(checkpoint);
            let events = runner.start(move |context| async move {
                let cfg = EventLogConfig::for_partition(restarted);
                let log = EventLog::open(context.child("restart"), cfg)
                    .await
                    .expect("a reopen after an interrupted reset must mount");
                log.replay().await.expect("replay").len()
            });
            assert!(
                events == 0 || events == 3,
                "a reset interrupted by a {label} failure left {events} event(s). A partial \
                 history is the outcome `EventLog::reset` exists to make impossible: no layer \
                 above can tell it from a legitimately short conversation"
            );
            outcomes.push((label, events));
        }

        // The matrix has to reach BOTH sides of the destructive boundary, or
        // it proves nothing about crossing it. Measured on this release: a
        // failed open, write, sync, scan, or torn write stops the reset before
        // its intent is durable, so the old history stands and the caller
        // retries; a failed resize, remove, or torn resize stops it after,
        // and the next open finishes the reset it staged.
        //
        // A reset that never staged intent would leave every case intact and
        // fail here, which is the point of asserting the distribution rather
        // than each row.
        assert!(
            outcomes.iter().any(|&(_, events)| events == 3),
            "no case stopped before the reset became durable: {outcomes:?}"
        );
        assert!(
            outcomes.iter().any(|&(_, events)| events == 0),
            "no case crossed the destructive boundary, so nothing here exercised recovery \
             FROM it: {outcomes:?}"
        );
    }

    /// Every state an interrupted reclaim can stop in still mounts as empty.
    ///
    /// `reclaim` is Commonware's final-teardown `destroy`, which is seven
    /// removals in a fixed order: the data blob, the data partition, the
    /// offsets blob, the offsets partition, then each of the two metadata
    /// copies and their partition. None of them shortens a file — a removal
    /// unlinks or it does not — so every interruption leaves a PREFIX of that
    /// order applied to a journal that was already empty.
    ///
    /// This walks all eight prefixes and requires each one to reopen as an
    /// empty journal. That is the property the erase path's retry depends on:
    /// it opens the partition before it removes anything, so a leftover that
    /// refused to mount would strand the erasure instead of finishing it.
    ///
    /// It holds today because `init` treats each missing piece as fresh — a
    /// missing data partition reads empty, a missing offsets partition reads
    /// empty, a missing checkpoint reads fresh — and because the reset left
    /// the recovery watermark at zero, so the one check that refuses a short
    /// offsets blob cannot fire.
    ///
    /// It is pinned rather than argued because it rests on `init_at_size` and
    /// `Partition::select`, both ALPHA. A release that tightens what `init`
    /// will mount turns this red instead of turning an erasure into a
    /// permanent refusal.
    #[test]
    fn every_prefix_of_a_reclaim_still_reopens_as_an_empty_journal() {
        use commonware_runtime::{Runner as _, tokio as cw_tokio};

        // The order Commonware removes them in.
        let order: [(&str, bool); 7] = [
            ("_data", false),
            ("_data", true),
            ("_offsets-blobs", false),
            ("_offsets-blobs", true),
            ("_offsets-metadata", false),
            ("_offsets-metadata", false),
            ("_offsets-metadata", true),
        ];

        for prefix in 0..=order.len() {
            let dir = std::env::temp_dir().join(format!(
                "polyc-eventlog-reclaim-prefix-{}-{prefix}",
                std::process::id()
            ));
            let _ = std::fs::remove_dir_all(&dir);
            std::fs::create_dir_all(&dir).expect("case directory");
            let cfg = EventLogConfig::for_partition("conv-reclaim-prefix");

            // A committed partition, then the reset that precedes every
            // reclaim. Anything left below is a leftover of an EMPTY journal,
            // which is the only state `reclaim` will act on.
            seed_three_events(&dir, &cfg);
            let reset_runner = cw_tokio::Runner::new(
                cw_tokio::Config::default().with_storage_directory(dir.clone()),
            );
            let reset_cfg = cfg.clone();
            reset_runner.start(move |context| async move {
                let log = EventLog::open(context.child("pre"), reset_cfg)
                    .await
                    .expect("open");
                let log = log.reset(context.child("pre")).await.expect("reset");
                assert_eq!(log.len().await, 0, "reclaim only ever runs on an empty log");
            });

            let mut removals = 0;
            for (suffix, whole_partition) in order.iter().take(prefix) {
                let path = dir.join(format!("conv-reclaim-prefix{suffix}"));
                if !path.exists() {
                    continue;
                }
                if *whole_partition {
                    std::fs::remove_dir_all(&path).expect("remove the partition directory");
                    removals += 1;
                } else if let Some(file) = std::fs::read_dir(&path)
                    .expect("partition directory")
                    .filter_map(Result::ok)
                    .map(|entry| entry.path())
                    .find(|path| path.is_file())
                {
                    std::fs::remove_file(file).expect("unlink one blob");
                    removals += 1;
                }
            }
            // Without this the case could pass by removing nothing at all,
            // which would assert only that a healthy partition reopens.
            assert_eq!(
                removals, prefix,
                "prefix {prefix} was meant to remove {prefix} artifact(s) and removed {removals}"
            );

            let outcome = reopen_and_count(&dir, &cfg);
            assert_eq!(
                outcome,
                Reopened::Holding(0),
                "a reclaim interrupted after {prefix} of its {} removals answered {outcome:?}; \
                 the erase path reopens the partition before it removes anything, so a leftover \
                 that will not mount strands the erasure instead of finishing it",
                order.len()
            );

            let _ = std::fs::remove_dir_all(&dir);
        }
    }
}