gemini_memory_rs/ingestion/
overlay.rs

1//! The session overlay — facts the user stated moments ago, usable now.
2//!
3//! Waiting for post-session reconciliation before a fact becomes retrievable
4//! would mean B forgets what it was told thirty seconds ago, which is the exact
5//! failure the whole system exists to avoid. The overlay closes that gap: it is
6//! searched alongside canonical memory, ranked above it, and presented hedged
7//! because it has not yet been reconciled.
8
9use chrono::{DateTime, Utc};
10
11use super::ledger::{SessionCandidate, SessionCandidateStatus};
12use crate::bm25::{IndexedMemory, MemoryIndex};
13use crate::core::{
14    CanonicalMemory, EvidenceCounters, MemoryId, MemorySource, MemoryStatus, MutationIntent,
15    PrivacyMetadata, RetrievalMetadata, SensitivityClass, SessionId, TemporalMetadata, UserId,
16    stable_hash,
17};
18
19/// The searchable projection of the session ledger.
20#[derive(Debug, Default)]
21pub struct SessionMemoryOverlay {
22    revision: u64,
23    index: MemoryIndex,
24}
25
26impl SessionMemoryOverlay {
27    /// An empty overlay.
28    pub fn new() -> Self {
29        Self::default()
30    }
31
32    /// Overlay revision, bumped on every rebuild that changed something.
33    pub fn revision(&self) -> u64 {
34        self.revision
35    }
36
37    /// The searchable index.
38    pub fn index(&self) -> &MemoryIndex {
39        &self.index
40    }
41
42    /// How many facts the overlay holds.
43    pub fn len(&self) -> usize {
44        self.index.len()
45    }
46
47    /// Whether the overlay is empty.
48    pub fn is_empty(&self) -> bool {
49        self.index.is_empty()
50    }
51
52    /// Rebuild from the ledger's usable candidates.
53    ///
54    /// Rebuilding wholesale rather than patching is what makes suppression
55    /// work: a candidate contradicted this turn simply stops being produced,
56    /// and cannot linger in the index asserting the old value.
57    pub fn rebuild(
58        &mut self,
59        owner: &UserId,
60        session_id: &SessionId,
61        candidates: &[SessionCandidate],
62        now: DateTime<Utc>,
63    ) -> usize {
64        let mut index = MemoryIndex::new();
65        for candidate in candidates {
66            if !candidate.status.is_usable() {
67                continue;
68            }
69            if candidate.status == SessionCandidateStatus::Suppressed {
70                continue;
71            }
72            // "forget that", "what do you remember" and friends are commands
73            // *about* memory. They belong in the ledger so consolidation can
74            // act on them, but they are not facts about the user and must
75            // never be handed back as recalled context.
76            if matches!(
77                candidate.mutation_intent,
78                Some(MutationIntent::List)
79                    | Some(MutationIntent::Forget)
80                    | Some(MutationIntent::Delete)
81            ) {
82                continue;
83            }
84            let provisional = provisional_memory(candidate, owner, session_id, now);
85            index.upsert(IndexedMemory::from_canonical(&provisional).as_session_overlay());
86        }
87        let changed = index.len() != self.index.len();
88        self.index = index;
89        if changed {
90            self.revision += 1;
91        } else {
92            // Content may have changed even when the count did not; the
93            // revision is a cache key, so err toward invalidating.
94            self.revision += 1;
95        }
96        self.index.len()
97    }
98
99    /// Drop every overlay fact, e.g. when a logical session ends.
100    pub fn clear(&mut self) {
101        self.index = MemoryIndex::new();
102        self.revision += 1;
103    }
104}
105
106/// A stable synthetic id for an uncommitted session fact.
107///
108/// Derived from the fingerprint so the same fact keeps the same id across
109/// rebuilds, and prefixed so it can never be mistaken for a canonical record.
110pub fn provisional_memory_id(candidate: &SessionCandidate) -> MemoryId {
111    MemoryId::new(format!(
112        "session_{}",
113        stable_hash(candidate.fingerprint.as_str())
114    ))
115}
116
117/// Project a session candidate as a provisional memory record.
118pub fn provisional_memory(
119    candidate: &SessionCandidate,
120    owner: &UserId,
121    session_id: &SessionId,
122    now: DateTime<Utc>,
123) -> CanonicalMemory {
124    let mut temporal = TemporalMetadata::created_at(now);
125    temporal.expires_at = candidate.expected_expiry;
126
127    CanonicalMemory {
128        id: provisional_memory_id(candidate),
129        owner: owner.clone(),
130        kind: candidate.kind,
131        predicate: candidate.predicate.clone(),
132        status: MemoryStatus::Staged,
133        confidence: candidate.confidence,
134        subject: candidate.subject.clone(),
135        value: candidate.value.clone(),
136        statement: candidate.canonical_statement.clone(),
137        evidence_summary: format!(
138            "Observed {} time(s) in the current session.",
139            candidate.evidence.len()
140        ),
141        source: MemorySource::from_explicitness(
142            candidate.explicitness,
143            session_id.clone(),
144            candidate.last_seen_turn,
145        ),
146        temporal,
147        retrieval: RetrievalMetadata {
148            subject: crate::core::normalize_token(&candidate.subject.display),
149            tags: derive_tags(candidate),
150            aliases: Vec::new(),
151            entities: candidate.subject.surface_forms(),
152            location: None,
153        },
154        evidence: EvidenceCounters {
155            count: candidate.evidence.len() as u32,
156            distinct_sessions: 1,
157            distinct_days: candidate.distinct_days().max(1),
158        },
159        privacy: PrivacyMetadata {
160            deletable: true,
161            exportable: true,
162            sensitivity: SensitivityClass::Normal,
163        },
164        temporal_scope: candidate.temporal_scope,
165        supersedes: Vec::new(),
166        superseded_by: None,
167        qualifier: None,
168    }
169}
170
171/// Tags for a provisional record: the predicate's parts plus the value's terms.
172fn derive_tags(candidate: &SessionCandidate) -> Vec<String> {
173    let mut tags: Vec<String> = candidate
174        .predicate
175        .as_str()
176        .split('_')
177        .map(str::to_string)
178        .collect();
179    tags.extend(crate::bm25::tokenize(&candidate.value.display()));
180    for term in &candidate.search_terms {
181        tags.extend(crate::bm25::tokenize(term));
182    }
183    tags.retain(|t| !t.is_empty());
184    tags.sort();
185    tags.dedup();
186    tags
187}
188
189#[cfg(test)]
190mod tests {
191    use super::*;
192    use crate::bm25::{MemoryOrigin, Query};
193    use crate::core::{
194        CanonicalPredicate, EntityRef, Explicitness, IngestionConfig, MemoryKind,
195        MemoryObservation, MemoryValue, ObservationId, ProposedPersistence, SpeakerAttribution,
196        TemporalScope, TranscriptEvidence, TurnId,
197    };
198    use crate::ingestion::ledger::{InMemorySessionLedger, SessionLedger};
199
200    fn observation(predicate: &str, value: &str, turn: u64) -> MemoryObservation {
201        MemoryObservation {
202            observation_id: ObservationId::generate(),
203            session_id: SessionId::new("ses_1"),
204            turn_id: TurnId(turn),
205            subject: EntityRef::user(),
206            predicate: CanonicalPredicate::new(predicate),
207            value: MemoryValue::Text(value.to_string()),
208            canonical_statement: format!("The user is {value}."),
209            kind: MemoryKind::Preference,
210            explicitness: Explicitness::ExplicitStatement,
211            confidence: 0.9,
212            persistence: ProposedPersistence::Durable,
213            temporal_scope: TemporalScope::Persistent,
214            valid_from: None,
215            expected_expiry: None,
216            transcript_evidence: TranscriptEvidence::new(format!("I am {value}")),
217            speaker_attribution: SpeakerAttribution::User,
218            sensitivity: SensitivityClass::Normal,
219            mutation_intent: None,
220            search_terms: Vec::new(),
221        }
222    }
223
224    async fn overlay_from(observations: Vec<MemoryObservation>) -> SessionMemoryOverlay {
225        let ledger =
226            InMemorySessionLedger::new(SessionId::new("ses_1"), IngestionConfig::default());
227        for obs in observations {
228            ledger.append_observation(obs).await.unwrap();
229        }
230        ledger.micro_reconcile();
231
232        let mut overlay = SessionMemoryOverlay::new();
233        overlay.rebuild(
234            &UserId::new("usr_1"),
235            &SessionId::new("ses_1"),
236            &ledger.usable_candidates(),
237            Utc::now(),
238        );
239        overlay
240    }
241
242    #[tokio::test]
243    async fn a_fact_stated_this_turn_is_searchable_immediately() {
244        let overlay = overlay_from(vec![observation("dietary_identity", "pescatarian", 1)]).await;
245        let hits = overlay
246            .index()
247            .search(&Query::new("pescatarian"), Utc::now());
248        assert_eq!(hits.len(), 1);
249        assert_eq!(hits[0].origin, MemoryOrigin::SessionOverlay);
250    }
251
252    #[tokio::test]
253    async fn a_suppressed_candidate_disappears_from_the_overlay() {
254        let overlay = overlay_from(vec![
255            observation("dietary_identity", "vegetarian", 1),
256            observation("dietary_identity", "pescatarian", 5),
257        ])
258        .await;
259
260        assert_eq!(overlay.len(), 1);
261        assert!(
262            overlay
263                .index()
264                .search(&Query::new("vegetarian"), Utc::now())
265                .is_empty()
266        );
267        assert!(
268            !overlay
269                .index()
270                .search(&Query::new("pescatarian"), Utc::now())
271                .is_empty()
272        );
273    }
274
275    #[tokio::test]
276    async fn provisional_ids_are_stable_across_rebuilds() {
277        let first = overlay_from(vec![observation("dietary_identity", "pescatarian", 1)]).await;
278        let second = overlay_from(vec![observation("dietary_identity", "pescatarian", 1)]).await;
279
280        let id_of = |o: &SessionMemoryOverlay| {
281            o.index().search(&Query::new("pescatarian"), Utc::now())[0]
282                .id
283                .clone()
284        };
285        assert_eq!(id_of(&first), id_of(&second));
286        assert!(id_of(&first).as_str().starts_with("session_"));
287    }
288
289    #[tokio::test]
290    async fn rebuilding_bumps_the_revision_so_caches_invalidate() {
291        let mut overlay = SessionMemoryOverlay::new();
292        let before = overlay.revision();
293        overlay.rebuild(
294            &UserId::new("usr_1"),
295            &SessionId::new("ses_1"),
296            &[],
297            Utc::now(),
298        );
299        assert!(overlay.revision() > before);
300    }
301
302    #[tokio::test]
303    async fn clearing_empties_the_overlay() {
304        let mut overlay =
305            overlay_from(vec![observation("dietary_identity", "pescatarian", 1)]).await;
306        overlay.clear();
307        assert!(overlay.is_empty());
308    }
309
310    #[tokio::test]
311    async fn a_provisional_record_is_staged_not_active() {
312        let ledger =
313            InMemorySessionLedger::new(SessionId::new("ses_1"), IngestionConfig::default());
314        ledger
315            .append_observation(observation("dietary_identity", "pescatarian", 1))
316            .await
317            .unwrap();
318        let candidate = &ledger.usable_candidates()[0];
319        let provisional = provisional_memory(
320            candidate,
321            &UserId::new("usr_1"),
322            &SessionId::new("ses_1"),
323            Utc::now(),
324        );
325        assert_eq!(provisional.status, MemoryStatus::Staged);
326        assert!(
327            provisional
328                .retrieval
329                .tags
330                .contains(&"pescatarian".to_string())
331        );
332    }
333}
gemini_memory_rs/ingestion/overlay.rs

gemini_memory_rs/ingestion/
overlay.rs
