polyc_eventlog/

lib.rs

//! 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, extend_and_sign, rebuild_from_events,
    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,
{
    /// 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>>,
}

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 page_cache = CacheRef::from_pooler(&context, config.page_size, config.page_cache_pages);
        let journal_cfg = variable::Config {
            partition: config.partition,
            items_per_section: config.items_per_section,
            compression: None,
            codec_config: config.event_cfg,
            page_cache,
            write_buffer: config.write_buffer,
        };
        let journal = variable::Journal::init(context, journal_cfg).await?;
        Ok(Self {
            journal: AsyncMutex::new(journal),
        })
    }

    /// Destroy the log: consume the handle and REMOVE the partition's
    /// underlying blobs (data + offsets) from storage. The erasure primitive
    /// (#216): after this, a fresh [`EventLog::open`] of the same partition
    /// starts empty.
    ///
    /// Deliberately consuming — a destroyed log has no valid further
    /// operation, and the caller must drop every other handle first (the
    /// control plane's host serializes this through its single command
    /// loop and evicts its cache entry before destroying).
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the underlying blob removal
    /// fails.
    pub async fn destroy(self) -> Result<(), EventLogError> {
        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());
        });
    }

    /// Events appended, committed, and re-opened from the same partition
    /// replay identically — persistence survives dropping the handle.
    /// Destroy removes the partition wholesale: a reopen starts empty, and
    /// appends after the reopen work normally (no resurrection of old
    /// events). The #216 erasure primitive.
    #[test]
    fn destroy_removes_the_partition_and_reopen_is_empty() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cfg = EventLogConfig::for_partition("destroy-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");
            log.destroy().await.expect("destroy");

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

    /// 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);
    }
}