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: per the prior `commonware-transport` spike, the
//! Commonware runtime cannot be started from inside a live tokio runtime, so a
//! tokio control plane hosts it on a dedicated thread. 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;
pub mod event;
pub mod integrity;
mod metrics;
pub mod nav;
pub mod taint;

pub use checkpoint::EventCountCheckpoint;
pub use error::EventLogError;
pub use event::{Event, EventCfg};
pub use integrity::{
    IntegrityError, MMR_SIGNED_ROOT_KIND, extend_and_sign, rebuild_from_events, verify_replay,
};
pub use taint::{
    GrantedCapabilities, TrifectaLegs, TrustTag, any_untrusted, any_untrusted_excluding,
    trifecta_legs,
};

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

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