agentyk_core/
event_log.rs

1//! Event log — the append-only record of everything a session did, and the
2//! persistence seam of the library.
3//!
4//! Sessions write through [`EventLog`] and can be resumed by replaying it
5//! (see [`crate::replay`]). Hosts choose durability by choosing an
6//! implementation: [`InMemoryEventLog`] ships here; `agentyk` adds a JSONL
7//! file log; a server host can implement this trait over a database.
8
9use std::collections::HashMap;
10use std::sync::Mutex;
11
12use async_trait::async_trait;
13use chrono::Utc;
14use serde::{Deserialize, Serialize};
15
16use crate::error::Result;
17use crate::event::{Event, EventData, EventRequest};
18use crate::id::{EventId, SessionId};
19
20/// Immutable address of a durable point in a session timeline.
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
22pub struct SessionPoint {
23    /// Session whose timeline is addressed.
24    pub session_id: SessionId,
25    /// Inclusive durable sequence, with zero denoting an empty session.
26    pub sequence: u64,
27}
28
29impl SessionPoint {
30    /// Address `sequence` in `session_id`.
31    pub fn new(session_id: SessionId, sequence: u64) -> Self {
32        Self {
33            session_id,
34            sequence,
35        }
36    }
37}
38
39/// Bounded forward read over one session timeline.
40#[derive(Debug, Clone, Copy, PartialEq, Eq)]
41pub struct EventRange {
42    /// Session to read.
43    pub session_id: SessionId,
44    /// Exclude events at or before this sequence.
45    pub after: Option<u64>,
46    /// Exclude events after this sequence.
47    pub through: Option<u64>,
48    /// Maximum events returned in one page.
49    pub limit: usize,
50}
51
52impl EventRange {
53    /// Read the first page of `session_id` with a conservative default size.
54    pub fn new(session_id: SessionId) -> Self {
55        Self {
56            session_id,
57            after: None,
58            through: None,
59            limit: 256,
60        }
61    }
62
63    /// Start strictly after `sequence`.
64    pub fn after(mut self, sequence: u64) -> Self {
65        self.after = Some(sequence);
66        self
67    }
68
69    /// Stop at `sequence`, inclusively.
70    pub fn through(mut self, sequence: u64) -> Self {
71        self.through = Some(sequence);
72        self
73    }
74
75    /// Return at most `limit` events. Zero returns an empty page.
76    pub fn limit(mut self, limit: usize) -> Self {
77        self.limit = limit;
78        self
79    }
80}
81
82/// One bounded event read plus cursors needed to continue consistently.
83#[derive(Debug, Clone, PartialEq)]
84pub struct EventPage {
85    /// Events in ascending durable sequence order.
86    pub events: Vec<Event>,
87    /// Cursor for the next page when more events remain in the requested range.
88    pub next: Option<SessionPoint>,
89    /// Stream head observed while serving this page.
90    pub head: SessionPoint,
91}
92
93/// Disposable projection anchored to an immutable session point.
94///
95/// Events remain authoritative. A snapshot only accelerates replay and may be
96/// dropped whenever its projection name or schema is no longer understood.
97#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
98pub struct ProjectionSnapshot {
99    /// Last event incorporated into the projection.
100    pub point: SessionPoint,
101    /// Stable projection family, such as `history` or `turn_state`.
102    pub projection: String,
103    /// Schema understood by the projection reducer.
104    pub schema_version: u32,
105    /// Projection-owned serialized state.
106    pub payload: serde_json::Value,
107}
108
109/// Optional storage for disposable replay accelerators.
110#[async_trait]
111pub trait SnapshotStore: Send + Sync {
112    /// Most recent compatible snapshot at or before `through`.
113    async fn load_latest(
114        &self,
115        session_id: SessionId,
116        projection: &str,
117        schema_version: u32,
118        through: Option<u64>,
119    ) -> Result<Option<ProjectionSnapshot>>;
120
121    /// Save or replace a snapshot at its immutable session point.
122    async fn save(&self, snapshot: ProjectionSnapshot) -> Result<()>;
123}
124
125/// In-memory snapshot storage for tests and short-lived applications.
126#[derive(Default)]
127pub struct InMemorySnapshotStore {
128    snapshots: Mutex<HashMap<(SessionId, String), Vec<ProjectionSnapshot>>>,
129}
130
131impl InMemorySnapshotStore {
132    /// Empty snapshot storage.
133    pub fn new() -> Self {
134        Self::default()
135    }
136}
137
138#[async_trait]
139impl SnapshotStore for InMemorySnapshotStore {
140    async fn load_latest(
141        &self,
142        session_id: SessionId,
143        projection: &str,
144        schema_version: u32,
145        through: Option<u64>,
146    ) -> Result<Option<ProjectionSnapshot>> {
147        let through = through.unwrap_or(u64::MAX);
148        let snapshots = self.snapshots.lock().expect("snapshot store poisoned");
149        Ok(snapshots
150            .get(&(session_id, projection.to_owned()))
151            .into_iter()
152            .flatten()
153            .filter(|snapshot| {
154                snapshot.schema_version == schema_version && snapshot.point.sequence <= through
155            })
156            .max_by_key(|snapshot| snapshot.point.sequence)
157            .cloned())
158    }
159
160    async fn save(&self, snapshot: ProjectionSnapshot) -> Result<()> {
161        let key = (snapshot.point.session_id, snapshot.projection.clone());
162        let mut snapshots = self.snapshots.lock().expect("snapshot store poisoned");
163        let entries = snapshots.entry(key).or_default();
164        if let Some(existing) = entries.iter_mut().find(|existing| {
165            existing.point.sequence == snapshot.point.sequence
166                && existing.schema_version == snapshot.schema_version
167        }) {
168            *existing = snapshot;
169        } else {
170            entries.push(snapshot);
171        }
172        Ok(())
173    }
174}
175
176/// Required stream version when appending a durable event batch.
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
178pub enum ExpectedVersion {
179    /// Append regardless of the current stream version.
180    Any,
181    /// Append only when the stream currently ends at this sequence.
182    Exact(u64),
183}
184
185/// Append-only, per-session-sequenced event storage.
186#[async_trait]
187pub trait EventStore: Send + Sync {
188    /// Atomically append a batch and assign ids, timestamps, and contiguous
189    /// per-session sequences.
190    ///
191    /// Every request must belong to `session_id`. An exact version prevents
192    /// two hosts from advancing the same session concurrently.
193    async fn append_batch(
194        &self,
195        session_id: SessionId,
196        expected: ExpectedVersion,
197        requests: Vec<EventRequest>,
198    ) -> Result<Vec<Event>>;
199
200    /// Read events after `sequence`, ordered by sequence.
201    ///
202    /// `None` reads the stream from its beginning.
203    async fn read_after(&self, session_id: SessionId, sequence: Option<u64>) -> Result<Vec<Event>>;
204
205    /// Current durable head without loading the stream body.
206    ///
207    /// Implementations should override this when they can answer from an
208    /// index. The default preserves compatibility for simple stores.
209    async fn head(&self, session_id: SessionId) -> Result<SessionPoint> {
210        let sequence = self
211            .read_after(session_id, None)
212            .await?
213            .last()
214            .and_then(|event| event.sequence)
215            .unwrap_or(0);
216        Ok(SessionPoint::new(session_id, sequence))
217    }
218
219    /// Read one bounded page. Implementations should push bounds into their
220    /// storage query; the default is correct for small custom stores.
221    async fn read_page(&self, range: EventRange) -> Result<EventPage> {
222        let head = self.head(range.session_id).await?;
223        let mut events = self.read_after(range.session_id, range.after).await?;
224        if let Some(through) = range.through {
225            events.retain(|event| event.sequence.is_some_and(|sequence| sequence <= through));
226        }
227        let has_more = events.len() > range.limit;
228        events.truncate(range.limit);
229        let next = has_more
230            .then(|| events.last().and_then(|event| event.sequence))
231            .flatten()
232            .map(|sequence| SessionPoint::new(range.session_id, sequence));
233        Ok(EventPage { events, next, head })
234    }
235
236    /// Read an immutable prefix through `point`, using bounded pages.
237    async fn read_through(&self, point: SessionPoint) -> Result<Vec<Event>> {
238        let head = self.head(point.session_id).await?;
239        if point.sequence > head.sequence {
240            return Err(crate::error::Error::EventLog(format!(
241                "point {} is past session head {}",
242                point.sequence, head.sequence
243            )));
244        }
245        let mut events = Vec::new();
246        let mut after = None;
247        loop {
248            let mut range = EventRange::new(point.session_id).through(point.sequence);
249            if let Some(sequence) = after {
250                range = range.after(sequence);
251            }
252            let page = self.read_page(range).await?;
253            events.extend(page.events);
254            match page.next {
255                Some(next) => after = Some(next.sequence),
256                None => return Ok(events),
257            }
258        }
259    }
260
261    /// Create a new session whose initial history is the prefix ending at
262    /// `origin`. The returned id is generated by the store.
263    ///
264    /// This default materializes the prefix. Database implementations may use
265    /// copy-on-write lineage as long as reads expose the same effective stream.
266    async fn fork(&self, origin: SessionPoint) -> Result<SessionId> {
267        let parent_head = self.head(origin.session_id).await?;
268        if origin.sequence > parent_head.sequence {
269            return Err(crate::error::Error::EventLog(format!(
270                "fork point {} is past session head {}",
271                origin.sequence, parent_head.sequence
272            )));
273        }
274        let child = SessionId::new();
275        let inherited = self.read_through(origin).await?.into_iter();
276        let mut requests: Vec<EventRequest> = inherited
277            .map(|event| EventRequest {
278                session_id: child,
279                turn_id: event.turn_id,
280                metadata: event.metadata,
281                data: event.data,
282            })
283            .collect();
284        requests.push(EventRequest::new(
285            child,
286            EventData::SessionForked {
287                parent_session_id: origin.session_id,
288                parent_sequence: origin.sequence,
289            },
290        ));
291        self.append_batch(child, ExpectedVersion::Exact(0), requests)
292            .await?;
293        Ok(child)
294    }
295
296    /// Append one event with optimistic concurrency against the current
297    /// stream version.
298    async fn append(&self, request: EventRequest) -> Result<Event> {
299        let session_id = request.session_id;
300        let version = self.head(session_id).await?.sequence;
301        self.append_batch(session_id, ExpectedVersion::Exact(version), vec![request])
302            .await?
303            .into_iter()
304            .next()
305            .ok_or_else(|| crate::error::Error::EventLog("append returned no event".into()))
306    }
307
308    /// All events for a session, ordered by sequence.
309    async fn read(&self, session_id: SessionId) -> Result<Vec<Event>> {
310        let head = self.head(session_id).await?;
311        self.read_through(head).await
312    }
313}
314
315/// In-memory event log. The default for `Agent::session()` and the natural
316/// choice for tests.
317#[derive(Default)]
318pub struct InMemoryEventLog {
319    sessions: Mutex<HashMap<SessionId, Vec<Event>>>,
320}
321
322impl InMemoryEventLog {
323    /// An empty log.
324    pub fn new() -> Self {
325        Self::default()
326    }
327}
328
329/// Compatibility name for [`EventStore`].
330pub use EventStore as EventLog;
331
332#[async_trait]
333impl EventStore for InMemoryEventLog {
334    async fn append_batch(
335        &self,
336        session_id: SessionId,
337        expected: ExpectedVersion,
338        requests: Vec<EventRequest>,
339    ) -> Result<Vec<Event>> {
340        let mut sessions = self.sessions.lock().expect("event log poisoned");
341        let stream = sessions.entry(session_id).or_default();
342        let actual = stream.len() as u64;
343        if let ExpectedVersion::Exact(expected) = expected
344            && expected != actual
345        {
346            return Err(crate::error::Error::EventConflict { expected, actual });
347        }
348        if requests
349            .iter()
350            .any(|request| request.session_id != session_id)
351        {
352            return Err(crate::error::Error::EventLog(
353                "event batch contains a different session".into(),
354            ));
355        }
356        let now = Utc::now();
357        let appended: Vec<Event> = requests
358            .into_iter()
359            .enumerate()
360            .map(|(index, request)| {
361                request.into_event(EventId::new(), now, actual + index as u64 + 1)
362            })
363            .collect();
364        stream.extend(appended.iter().cloned());
365        Ok(appended)
366    }
367
368    async fn read_after(&self, session_id: SessionId, sequence: Option<u64>) -> Result<Vec<Event>> {
369        let sessions = self.sessions.lock().expect("event log poisoned");
370        let after = sequence.unwrap_or(0);
371        Ok(sessions
372            .get(&session_id)
373            .into_iter()
374            .flatten()
375            .filter(|event| event.sequence.is_some_and(|value| value > after))
376            .cloned()
377            .collect())
378    }
379
380    async fn head(&self, session_id: SessionId) -> Result<SessionPoint> {
381        let sessions = self.sessions.lock().expect("event log poisoned");
382        let sequence = sessions
383            .get(&session_id)
384            .and_then(|events| events.last())
385            .and_then(|event| event.sequence)
386            .unwrap_or(0);
387        Ok(SessionPoint::new(session_id, sequence))
388    }
389
390    async fn read_page(&self, range: EventRange) -> Result<EventPage> {
391        let sessions = self.sessions.lock().expect("event log poisoned");
392        let stream = sessions.get(&range.session_id);
393        let head_sequence = stream
394            .and_then(|events| events.last())
395            .and_then(|event| event.sequence)
396            .unwrap_or(0);
397        let after = range.after.unwrap_or(0);
398        let through = range.through.unwrap_or(u64::MAX);
399        let mut matching = stream.into_iter().flatten().filter(|event| {
400            event
401                .sequence
402                .is_some_and(|sequence| sequence > after && sequence <= through)
403        });
404        let events: Vec<Event> = matching.by_ref().take(range.limit).cloned().collect();
405        let next = matching
406            .next()
407            .and_then(|_| events.last())
408            .and_then(|event| event.sequence)
409            .map(|sequence| SessionPoint::new(range.session_id, sequence));
410        Ok(EventPage {
411            events,
412            next,
413            head: SessionPoint::new(range.session_id, head_sequence),
414        })
415    }
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421    use crate::event::EventData;
422
423    fn block_on<F: std::future::Future>(future: F) -> F::Output {
424        use std::pin::pin;
425        use std::task::{Context, Poll, Waker};
426
427        let mut future = pin!(future);
428        let waker = Waker::noop();
429        let mut context = Context::from_waker(waker);
430        loop {
431            if let Poll::Ready(output) = future.as_mut().poll(&mut context) {
432                return output;
433            }
434        }
435    }
436
437    #[test]
438    fn batch_append_is_atomic_and_version_checked() {
439        block_on(async {
440            let log = InMemoryEventLog::new();
441            let session_id = SessionId::new();
442            let requests = vec![
443                EventRequest::new(session_id, EventData::TurnStarted { max_iterations: 4 }),
444                EventRequest::new(session_id, EventData::TurnCancelled),
445            ];
446            let appended = log
447                .append_batch(session_id, ExpectedVersion::Exact(0), requests)
448                .await
449                .unwrap();
450            assert_eq!(appended.len(), 2);
451            assert_eq!(appended[0].sequence, Some(1));
452            assert_eq!(appended[1].sequence, Some(2));
453
454            let error = log
455                .append_batch(
456                    session_id,
457                    ExpectedVersion::Exact(0),
458                    vec![EventRequest::new(session_id, EventData::TurnCancelled)],
459                )
460                .await
461                .unwrap_err();
462            assert!(matches!(
463                error,
464                crate::error::Error::EventConflict {
465                    expected: 0,
466                    actual: 2
467                }
468            ));
469            assert_eq!(log.read_after(session_id, Some(1)).await.unwrap().len(), 1);
470        });
471    }
472}
agentyk_core/event_log.rs

agentyk_core/
event_log.rs
