gemini_memory_rs/core/
events.rs

1//! The append-only memory event log — the recovery and audit backbone.
2//!
3//! Everything durable that happens to memory is an event first and a mutation
4//! second. A crash mid-session loses the in-process ledger but not the events,
5//! so the session overlay can be rebuilt by replay.
6
7use async_trait::async_trait;
8use chrono::{DateTime, Utc};
9use serde::{Deserialize, Serialize};
10use std::sync::Arc;
11
12use super::domain::{FactFingerprint, MemoryObservation, MutationIntent, stable_hash};
13use super::error::MemoryError;
14use super::ids::{EventId, MemoryId, SessionId, TurnId, UserId};
15use super::policy::DiscardReason;
16
17/// A durable fact about something that happened to memory.
18#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
19#[serde(tag = "event", rename_all = "snake_case")]
20pub enum MemoryEvent {
21    /// A finalized user utterance was accepted as evidence.
22    FinalTranscriptRecorded {
23        /// The finalized text.
24        text: String,
25    },
26    /// An observation was extracted from a finalized utterance.
27    ObservationExtracted {
28        /// The observation.
29        observation: Box<MemoryObservation>,
30    },
31    /// An extraction attempt failed outright.
32    ///
33    /// Distinct from "the turn revealed nothing": a systematically failing
34    /// extractor and a quiet conversation look identical from the ledger, and
35    /// only this event tells them apart.
36    ExtractionFailed {
37        /// Which extraction stage failed.
38        stage: String,
39        /// Why.
40        reason: String,
41    },
42    /// An observation was refused by policy.
43    ObservationRejected {
44        /// Fingerprint of the refused candidate, for auditing.
45        fingerprint: FactFingerprint,
46        /// Why it was refused.
47        reason: DiscardReason,
48    },
49    /// Two observations were recognised as the same candidate.
50    SessionCandidateMerged {
51        /// The surviving fingerprint.
52        fingerprint: FactFingerprint,
53        /// Evidence count after the merge.
54        evidence_count: u32,
55    },
56    /// The in-session overlay changed.
57    SessionOverlayUpdated {
58        /// Overlay revision after the change.
59        revision: u64,
60    },
61    /// The user issued an explicit memory command.
62    ExplicitMutationRequested {
63        /// What they asked for.
64        intent: MutationIntent,
65        /// The statement they gave, verbatim.
66        statement: String,
67    },
68    /// A long session reached a checkpoint.
69    SessionCheckpointed {
70        /// Turns completed at checkpoint time.
71        turns: u64,
72    },
73    /// A logical session was sealed and accepts no further writes.
74    SessionSealed {
75        /// Candidates carried into consolidation.
76        candidate_count: usize,
77    },
78    /// Consolidation proposed a mutation.
79    MutationProposed {
80        /// Fingerprint of the proposal.
81        fingerprint: FactFingerprint,
82        /// The resolution kind, as a stable label.
83        kind: String,
84    },
85    /// A mutation was committed to canonical memory.
86    MutationCommitted {
87        /// The affected record.
88        memory_id: MemoryId,
89        /// The resolution kind, as a stable label.
90        kind: String,
91    },
92    /// A proposal was refused at commit time.
93    MutationRejected {
94        /// Fingerprint of the refused proposal.
95        fingerprint: FactFingerprint,
96        /// Why it was refused.
97        reason: DiscardReason,
98    },
99    /// An active record was replaced.
100    MemorySuperseded {
101        /// The record that was replaced.
102        old: MemoryId,
103        /// The record that replaced it.
104        new: MemoryId,
105    },
106    /// A record was deleted at the user's request.
107    MemoryDeleted {
108        /// The removed record.
109        memory_id: MemoryId,
110    },
111    /// A new retrieval index revision was published.
112    IndexRevisionPublished {
113        /// The revision that is now serving.
114        revision: u64,
115    },
116}
117
118impl MemoryEvent {
119    /// A stable label for metrics and log filtering.
120    pub fn label(&self) -> &'static str {
121        match self {
122            Self::FinalTranscriptRecorded { .. } => "final_transcript_recorded",
123            Self::ObservationExtracted { .. } => "observation_extracted",
124            Self::ExtractionFailed { .. } => "extraction_failed",
125            Self::ObservationRejected { .. } => "observation_rejected",
126            Self::SessionCandidateMerged { .. } => "session_candidate_merged",
127            Self::SessionOverlayUpdated { .. } => "session_overlay_updated",
128            Self::ExplicitMutationRequested { .. } => "explicit_mutation_requested",
129            Self::SessionCheckpointed { .. } => "session_checkpointed",
130            Self::SessionSealed { .. } => "session_sealed",
131            Self::MutationProposed { .. } => "mutation_proposed",
132            Self::MutationCommitted { .. } => "mutation_committed",
133            Self::MutationRejected { .. } => "mutation_rejected",
134            Self::MemorySuperseded { .. } => "memory_superseded",
135            Self::MemoryDeleted { .. } => "memory_deleted",
136            Self::IndexRevisionPublished { .. } => "index_revision_published",
137        }
138    }
139}
140
141/// The current event schema version. Bumped whenever [`MemoryEvent`] changes
142/// shape in a way replay must account for.
143pub const EVENT_SCHEMA_VERSION: u32 = 1;
144
145/// An event plus the addressing metadata every consumer needs.
146#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
147pub struct MemoryEventEnvelope {
148    /// Identifier for this envelope.
149    pub event_id: EventId,
150    /// When it occurred.
151    pub occurred_at: DateTime<Utc>,
152    /// Whose memory it concerns.
153    pub user_id: UserId,
154    /// The logical session it belongs to.
155    pub logical_session_id: SessionId,
156    /// The turn it belongs to, when applicable.
157    pub turn_id: Option<TurnId>,
158    /// Deduplication key for at-least-once delivery.
159    pub idempotency_key: String,
160    /// Schema version of the payload.
161    pub schema_version: u32,
162    /// The event itself.
163    pub payload: MemoryEvent,
164}
165
166impl MemoryEventEnvelope {
167    /// Wrap an event for a user and session.
168    ///
169    /// The idempotency key is derived from the addressing tuple and the payload
170    /// so a retried append is recognised as a duplicate rather than duplicated.
171    pub fn new(
172        user_id: UserId,
173        logical_session_id: SessionId,
174        turn_id: Option<TurnId>,
175        payload: MemoryEvent,
176        now: DateTime<Utc>,
177    ) -> Self {
178        let payload_repr = serde_json::to_string(&payload).unwrap_or_default();
179        let idempotency_key = stable_hash(&format!(
180            "{user_id}|{logical_session_id}|{}|{}",
181            turn_id.map(|t| t.0).unwrap_or_default(),
182            payload_repr
183        ));
184        Self {
185            event_id: EventId::generate(),
186            occurred_at: now,
187            user_id,
188            logical_session_id,
189            turn_id,
190            idempotency_key,
191            schema_version: EVENT_SCHEMA_VERSION,
192            payload,
193        }
194    }
195}
196
197/// A durable append-only sink for memory events.
198///
199/// The live session worker awaits *acceptance* here — not extraction, not
200/// reconciliation. That is the only point where the engine may tell a user
201/// their correction was durably recorded.
202#[async_trait]
203pub trait MemoryEventLog: Send + Sync {
204    /// Append an envelope, returning once it is durable.
205    async fn append(&self, envelope: MemoryEventEnvelope) -> Result<(), MemoryError>;
206
207    /// Read back every event for a logical session, in append order.
208    async fn replay_session(
209        &self,
210        session_id: &SessionId,
211    ) -> Result<Vec<MemoryEventEnvelope>, MemoryError>;
212}
213
214/// An in-process event log, used by tests and single-node deployments.
215///
216/// Deduplicates on `idempotency_key`, so replaying an append is a no-op.
217#[derive(Debug, Default)]
218pub struct InMemoryEventLog {
219    entries: parking_lot::RwLock<Vec<MemoryEventEnvelope>>,
220    seen: parking_lot::RwLock<std::collections::HashSet<String>>,
221}
222
223impl InMemoryEventLog {
224    /// An empty log.
225    pub fn new() -> Self {
226        Self::default()
227    }
228
229    /// Every event appended so far, in order.
230    pub fn entries(&self) -> Vec<MemoryEventEnvelope> {
231        self.entries.read().clone()
232    }
233
234    /// How many events have been appended.
235    pub fn len(&self) -> usize {
236        self.entries.read().len()
237    }
238
239    /// Whether the log is empty.
240    pub fn is_empty(&self) -> bool {
241        self.entries.read().is_empty()
242    }
243
244    /// Count of appended events carrying a given label.
245    pub fn count_label(&self, label: &str) -> usize {
246        self.entries
247            .read()
248            .iter()
249            .filter(|e| e.payload.label() == label)
250            .count()
251    }
252}
253
254#[async_trait]
255impl MemoryEventLog for InMemoryEventLog {
256    async fn append(&self, envelope: MemoryEventEnvelope) -> Result<(), MemoryError> {
257        let mut seen = self.seen.write();
258        if !seen.insert(envelope.idempotency_key.clone()) {
259            return Ok(());
260        }
261        drop(seen);
262        self.entries.write().push(envelope);
263        Ok(())
264    }
265
266    async fn replay_session(
267        &self,
268        session_id: &SessionId,
269    ) -> Result<Vec<MemoryEventEnvelope>, MemoryError> {
270        Ok(self
271            .entries
272            .read()
273            .iter()
274            .filter(|e| &e.logical_session_id == session_id)
275            .cloned()
276            .collect())
277    }
278}
279
280/// A convenience wrapper that stamps the user and session onto every append.
281#[derive(Clone)]
282pub struct SessionEventWriter {
283    log: Arc<dyn MemoryEventLog>,
284    user_id: UserId,
285    session_id: SessionId,
286}
287
288impl SessionEventWriter {
289    /// Bind a log to one user and logical session.
290    pub fn new(log: Arc<dyn MemoryEventLog>, user_id: UserId, session_id: SessionId) -> Self {
291        Self {
292            log,
293            user_id,
294            session_id,
295        }
296    }
297
298    /// Append an event for a specific turn.
299    pub async fn append(
300        &self,
301        turn_id: Option<TurnId>,
302        payload: MemoryEvent,
303    ) -> Result<(), MemoryError> {
304        let envelope = MemoryEventEnvelope::new(
305            self.user_id.clone(),
306            self.session_id.clone(),
307            turn_id,
308            payload,
309            Utc::now(),
310        );
311        self.log.append(envelope).await
312    }
313
314    /// The bound user.
315    pub fn user_id(&self) -> &UserId {
316        &self.user_id
317    }
318
319    /// The bound logical session.
320    pub fn session_id(&self) -> &SessionId {
321        &self.session_id
322    }
323}
324
325/// The result of applying one memory transaction.
326#[derive(Debug, Clone, Default, PartialEq, Eq)]
327pub struct CommitReceipt {
328    /// The repository revision after the commit.
329    pub revision: u64,
330    /// Records created or updated.
331    pub written: Vec<MemoryId>,
332    /// Records removed.
333    pub deleted: Vec<MemoryId>,
334}
335
336#[cfg(test)]
337mod tests {
338    use super::*;
339
340    fn envelope(payload: MemoryEvent) -> MemoryEventEnvelope {
341        MemoryEventEnvelope::new(
342            UserId::new("usr_1"),
343            SessionId::new("ses_1"),
344            Some(TurnId(1)),
345            payload,
346            Utc::now(),
347        )
348    }
349
350    #[tokio::test]
351    async fn appends_are_idempotent_on_replay() {
352        let log = InMemoryEventLog::new();
353        let e = envelope(MemoryEvent::FinalTranscriptRecorded {
354            text: "hello".into(),
355        });
356        log.append(e.clone()).await.unwrap();
357        log.append(e).await.unwrap();
358        assert_eq!(log.len(), 1);
359    }
360
361    #[tokio::test]
362    async fn distinct_payloads_are_distinct_events() {
363        let log = InMemoryEventLog::new();
364        log.append(envelope(MemoryEvent::FinalTranscriptRecorded {
365            text: "a".into(),
366        }))
367        .await
368        .unwrap();
369        log.append(envelope(MemoryEvent::FinalTranscriptRecorded {
370            text: "b".into(),
371        }))
372        .await
373        .unwrap();
374        assert_eq!(log.len(), 2);
375    }
376
377    #[tokio::test]
378    async fn replay_is_scoped_to_one_session() {
379        let log = InMemoryEventLog::new();
380        log.append(envelope(MemoryEvent::SessionSealed { candidate_count: 1 }))
381            .await
382            .unwrap();
383        log.append(MemoryEventEnvelope::new(
384            UserId::new("usr_1"),
385            SessionId::new("ses_other"),
386            None,
387            MemoryEvent::SessionSealed { candidate_count: 2 },
388            Utc::now(),
389        ))
390        .await
391        .unwrap();
392
393        let replayed = log.replay_session(&SessionId::new("ses_1")).await.unwrap();
394        assert_eq!(replayed.len(), 1);
395    }
396}
gemini_memory_rs/core/events.rs

gemini_memory_rs/core/
events.rs
