writerslogic_warm/core/memory_api/
mod.rs

1// Copyright 2024-2026 WritersLogic Contributors
2// SPDX-License-Identifier: Apache-2.0
3
4//! A high-level memory API over the document engine: `remember`, `recall`, `forget` and
5//! `explain`, with temporal validity, supersession, and opt-in consolidation and forgetting.
6//!
7//! # Design
8//!
9//! A fact is an ordinary document. Its memory fields live in one reserved metadata key
10//! (`_mem`, see `stamp`), so the API adds no persisted structure: stores written before it
11//! open unchanged, and journaling, replay and encryption apply as for any document. The
12//! engine's documents, vectors and raw search stay reachable through [`Memory::core`];
13//! nothing here is required to use them.
14//!
15//! Two time axes are kept apart. Valid time is when a fact holds (`valid_from`, `valid_to`,
16//! and the event time it describes); knowledge time is when the store ingested it. Recall
17//! evaluates validity "as of" a time and, optionally, what was `known_at` an ingest time.
18//!
19//! Supersession links are written once, on the superseding fact, and never edit the old one,
20//! so history is never rewritten or lost. The reverse links live in a derived in-memory
21//! index that is rebuilt from the documents whenever the store changes behind the API's back.
22//!
23//! Every policy is a trait with a default that changes nothing: supersession is explicit,
24//! recency does not rescale scores, nothing is forgotten, nothing is consolidated. Each has
25//! a serde spec in [`MemoryConfig`] for bindings and may be replaced with any implementation
26//! through the `with_*` builders.
27
28/// Fails with [`MemoryError::Invalid`] unless the condition holds.
29macro_rules! check {
30    ($cond:expr, $($arg:tt)+) => {
31        if !($cond) {
32            return Err(anyhow::Error::new($crate::core::memory_api::MemoryError::Invalid(
33                format!($($arg)+),
34            )));
35        }
36    };
37}
38
39/// Fails with [`MemoryError::Invalid`].
40macro_rules! invalid {
41    ($($arg:tt)+) => {
42        return Err(anyhow::Error::new($crate::core::memory_api::MemoryError::Invalid(
43            format!($($arg)+),
44        )))
45    };
46}
47
48#[cfg(feature = "server")]
49pub mod contract;
50#[cfg(feature = "conversation-memory")]
51mod conversation;
52mod index;
53mod json;
54mod maintenance;
55mod policy;
56mod recall;
57mod stamp;
58mod types;
59mod write;
60
61use crate::core::sync::RwLock;
62use anyhow::Result;
63use serde::{Deserialize, Serialize};
64use std::sync::Arc;
65
66use crate::core::WarmCore;
67use index::Index;
68
69#[cfg(feature = "conversation-memory")]
70pub use conversation::*;
71pub use json::OPERATIONS;
72pub use policy::*;
73pub use types::*;
74
75/// An error the caller can act on. Anything that is not one of these (journal I/O, a
76/// poisoned invariant) reaches the caller as a plain `anyhow` error; downcast with
77/// `error.downcast_ref::<MemoryError>()` to tell the two apart.
78#[derive(Debug, thiserror::Error)]
79pub enum MemoryError {
80    /// The request is malformed or breaks a rule (bad field, unknown superseded id, cycle).
81    #[error("{0}")]
82    Invalid(String),
83    /// The named fact does not exist.
84    #[error("{0}")]
85    NotFound(String),
86    /// The id exists with different content and `onIdConflict` is `reject`.
87    #[error("{0}")]
88    Conflict(String),
89    /// Concurrent writes prevented a coherent plan or recall; retry the operation.
90    #[error("{0}")]
91    Concurrent(String),
92}
93
94/// What `remember` does when an id exists with different content.
95#[cfg_attr(feature = "server", derive(schemars::JsonSchema))]
96#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
97#[serde(rename_all = "kebab-case")]
98pub enum IdConflict {
99    /// Fail. Default: nothing is overwritten by accident.
100    #[default]
101    Reject,
102    /// Overwrite the stored fact. This discards its previous text; use supersession to keep
103    /// history instead.
104    Replace,
105}
106
107/// Plain-data configuration; every field has a default that changes no behavior.
108#[cfg_attr(feature = "server", derive(schemars::JsonSchema))]
109#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
110#[serde(rename_all = "camelCase", default, deny_unknown_fields)]
111pub struct MemoryConfig {
112    /// Results per recall when `k` is absent.
113    pub default_k: u32,
114    /// Which facts recall returns when `history` is absent.
115    pub default_history: HistoryMode,
116    pub supersession: SupersessionSpec,
117    pub recency: RecencySpec,
118    /// Multiplier on the score of non-current facts when recall includes them.
119    pub historical_weight: f64,
120    pub forgetting: ForgettingSpec,
121    pub consolidation: ConsolidationSpec,
122    pub on_id_conflict: IdConflict,
123}
124
125impl Default for MemoryConfig {
126    fn default() -> Self {
127        Self {
128            default_k: 10,
129            default_history: HistoryMode::Current,
130            supersession: SupersessionSpec::Explicit,
131            recency: RecencySpec::None,
132            historical_weight: 1.0,
133            forgetting: ForgettingSpec::KeepAll,
134            consolidation: ConsolidationSpec::None,
135            on_id_conflict: IdConflict::Reject,
136        }
137    }
138}
139
140impl MemoryConfig {
141    pub fn validate(&self) -> Result<()> {
142        check!(
143            (1..=1000).contains(&self.default_k),
144            "defaultK must be 1..=1000"
145        );
146        check!(
147            self.historical_weight.is_finite() && self.historical_weight >= 0.0,
148            "historicalWeight must be finite and nonnegative"
149        );
150        if let RecencySpec::Exponential {
151            half_life_ms,
152            floor,
153            ..
154        } = self.recency
155        {
156            check!(half_life_ms > 0, "recency halfLifeMs must be positive");
157            check!(
158                (0.0..=1.0).contains(&floor),
159                "recency floor must be in [0, 1]"
160            );
161        }
162        if let ForgettingSpec::MaxFacts { limit } = self.forgetting {
163            check!(limit > 0, "forgetting limit must be positive");
164        }
165        Ok(())
166    }
167}
168
169/// The memory API over one engine.
170///
171/// Safe to share across threads. Policy plans use revision-conditional commits; reads retry
172/// when their document and policy snapshots differ. Persistent commits are serialized.
173/// Several `Memory` values over one core stay correct: each notices the others' writes by
174/// the store revision and rebuilds its index, at the cost of one scan.
175pub struct Memory {
176    core: Arc<WarmCore>,
177    config: MemoryConfig,
178    supersession: Arc<dyn SupersessionPolicy>,
179    /// Whether the supersession policy can return links; the default never does, so writes
180    /// skip building its candidates.
181    supersession_active: bool,
182    recency: Arc<dyn RecencyPolicy>,
183    forgetting: Arc<dyn ForgettingPolicy>,
184    consolidation: Arc<dyn ConsolidationPolicy>,
185    entities: Option<Arc<dyn EntityExtractor>>,
186    observer: Option<Arc<dyn MemoryObserver>>,
187    clock: Arc<dyn Clock>,
188    index: RwLock<Index>,
189}
190
191impl Memory {
192    /// A memory over `core` with the default, do-nothing policies.
193    pub fn new(core: Arc<WarmCore>) -> Self {
194        Self::with_config(core, MemoryConfig::default()).expect("the default config is valid")
195    }
196
197    pub fn with_config(core: Arc<WarmCore>, config: MemoryConfig) -> Result<Self> {
198        config.validate()?;
199        let supersession_active = config.supersession != SupersessionSpec::Explicit;
200        let supersession: Arc<dyn SupersessionPolicy> = match config.supersession {
201            SupersessionSpec::Explicit => Arc::new(ExplicitOnly),
202            SupersessionSpec::LatestBySubject => Arc::new(LatestBySubject),
203        };
204        let recency: Arc<dyn RecencyPolicy> = match config.recency {
205            RecencySpec::None => Arc::new(NoRecency),
206            RecencySpec::Exponential {
207                half_life_ms,
208                basis,
209                floor,
210            } => Arc::new(ExponentialRecency {
211                half_life_ms,
212                basis,
213                floor,
214            }),
215        };
216        let forgetting: Arc<dyn ForgettingPolicy> = match config.forgetting {
217            ForgettingSpec::KeepAll => Arc::new(KeepAll),
218            ForgettingSpec::ExpiredFor { grace_ms } => Arc::new(ExpiredFor { grace_ms }),
219            ForgettingSpec::SupersededFor { age_ms } => Arc::new(SupersededFor { age_ms }),
220            ForgettingSpec::MaxFacts { limit } => Arc::new(MaxFacts { limit }),
221        };
222        let consolidation: Arc<dyn ConsolidationPolicy> = match config.consolidation {
223            ConsolidationSpec::None => Arc::new(NoConsolidation),
224            ConsolidationSpec::CollapseDuplicates => Arc::new(CollapseDuplicates),
225        };
226        Ok(Self {
227            core,
228            config,
229            supersession,
230            supersession_active,
231            recency,
232            forgetting,
233            consolidation,
234            entities: None,
235            observer: None,
236            clock: Arc::new(SystemClock),
237            index: RwLock::new(Index::default()),
238        })
239    }
240
241    /// Replaces the supersession policy.
242    pub fn with_supersession(mut self, policy: Arc<dyn SupersessionPolicy>) -> Self {
243        self.supersession = policy;
244        self.supersession_active = true;
245        self
246    }
247
248    pub fn with_recency(mut self, policy: Arc<dyn RecencyPolicy>) -> Self {
249        self.recency = policy;
250        self
251    }
252
253    pub fn with_forgetting(mut self, policy: Arc<dyn ForgettingPolicy>) -> Self {
254        self.forgetting = policy;
255        self
256    }
257
258    pub fn with_consolidation(mut self, policy: Arc<dyn ConsolidationPolicy>) -> Self {
259        self.consolidation = policy;
260        self
261    }
262
263    pub fn with_entity_extractor(mut self, extractor: Arc<dyn EntityExtractor>) -> Self {
264        self.entities = Some(extractor);
265        self
266    }
267
268    pub fn with_observer(mut self, observer: Arc<dyn MemoryObserver>) -> Self {
269        self.observer = Some(observer);
270        self
271    }
272
273    /// Replaces the clock, for deterministic ingest times.
274    pub fn with_clock(mut self, clock: Arc<dyn Clock>) -> Self {
275        self.clock = clock;
276        self
277    }
278
279    pub fn config(&self) -> &MemoryConfig {
280        &self.config
281    }
282
283    /// The engine underneath, for documents, vectors and raw search. Facts are documents
284    /// with the fact id as the document id.
285    pub fn core(&self) -> &Arc<WarmCore> {
286        &self.core
287    }
288
289    fn with_snapshot<T>(&self, mut read: impl FnMut(&Index) -> Result<T>) -> Result<T> {
290        for _ in 0..16 {
291            let Some(revision) = self.planning_revision() else {
292                continue;
293            };
294            let result = {
295                let index = self.index.read();
296                if index.revision != Some(revision) {
297                    continue;
298                }
299                read(&index)
300            };
301            if self.core.memory_revision_matches(revision) {
302                return result;
303            }
304        }
305        Err(MemoryError::Concurrent(
306            "memory changed during the snapshot; retry the operation".into(),
307        )
308        .into())
309    }
310
311    fn with_revision<T>(&self, mut read: impl FnMut() -> Result<T>) -> Result<T> {
312        for _ in 0..16 {
313            let Some(revision) = self.planning_revision() else {
314                continue;
315            };
316            let result = read();
317            if self.core.memory_revision_matches(revision) {
318                return result;
319            }
320        }
321        Err(MemoryError::Concurrent(
322            "memory changed while assembling the result; retry the operation".into(),
323        )
324        .into())
325    }
326
327    fn planning_revision(&self) -> Option<u64> {
328        self.ensure_fresh();
329        let revision = self.core.revision();
330        (self.index.read().revision == Some(revision)).then_some(revision)
331    }
332
333    /// Brings the index up to the store's current revision.
334    fn ensure_fresh(&self) {
335        let revision = self.core.revision();
336        if self.index.read().revision == Some(revision) {
337            return;
338        }
339        let mut index = self.index.write();
340        if index.revision == Some(self.core.revision()) {
341            return;
342        }
343        *index = Index::rebuild(&self.core);
344    }
345
346    /// Records that this process's own commit moved the store from `before` by exactly
347    /// `commits` revisions and the index already reflects it; anything else (another writer
348    /// interleaved, or the change was not tracked) leaves the index to rebuild. `apply`
349    /// returns false, before changing the index, when it cannot track the change.
350    fn settle(&self, before: u64, commits: u64, apply: impl FnOnce(&mut Index) -> bool) {
351        let after = self.core.revision();
352        let mut index = self.index.write();
353        index.revision =
354            (index.revision == Some(before) && after == before + commits && apply(&mut index))
355                .then_some(after);
356    }
357}
358
359#[cfg(test)]
360mod tests;
writerslogic_warm/core/memory_api/mod.rs

writerslogic_warm/core/memory_api/
mod.rs
