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 error;
pub mod event;
pub mod nav;
pub mod taint;

pub use error::EventLogError;
pub use event::{Event, EventCfg};
pub use taint::{
    GrantedCapabilities, TrifectaLegs, TrustTag, any_untrusted, any_untrusted_excluding,
    trifecta_legs,
};

use commonware_runtime::buffer::paged::CacheRef;
use commonware_storage::journal::contiguous::{Reader as _, variable};
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);

/// 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,
{
    journal: 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 })
    }

    /// 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.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`: the underlying journal serializes writes internally via
    /// interior mutability, so a shared reference suffices.
    ///
    /// 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> {
        Ok(self.journal.append(event).await?)
    }

    /// 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.size().await
    }

    /// 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.
    // The replay stream borrows the `reader` guard for its whole lifetime, so
    // the guard cannot be dropped before the stream is consumed — the lint's
    // suggested early drop would not compile here.
    #[allow(clippy::significant_drop_tightening)]
    pub async fn replay_with_positions(&self) -> Result<Vec<(u64, Event)>, EventLogError> {
        let reader = self.journal.reader().await;
        let start = reader.bounds().start;
        let stream = reader.replay(REPLAY_BUFFER, start).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.
    // The replay stream borrows the `reader` guard for its whole lifetime (see
    // [`EventLog::replay_with_positions`]); the lint's suggested early drop
    // would not compile.
    #[allow(clippy::significant_drop_tightening)]
    pub async fn replay_from_with_positions(
        &self,
        start: u64,
    ) -> Result<Vec<(u64, Event)>, EventLogError> {
        let reader = self.journal.reader().await;
        let bounds = reader.bounds();
        let from = start.max(bounds.start);
        if from >= bounds.end {
            return Ok(Vec::new());
        }
        let stream = reader.replay(REPLAY_BUFFER, from).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())
    }

    /// Durably persist all buffered appends, guaranteeing they survive a crash.
    ///
    /// # Errors
    ///
    /// Returns [`EventLogError::Journal`] if the underlying flush fails.
    pub async fn commit(&self) -> Result<(), EventLogError> {
        Ok(self.journal.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.sync().await?)
    }
}

#[cfg(test)]
mod tests {
    use super::{Event, EventLog, EventLogConfig};
    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 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");
        });
    }

    #[test]
    fn reopen_recovers_committed_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());
        });
    }

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