writerslogic_warm/core/memory_api/
maintenance.rs

1// Copyright 2024-2026 WritersLogic Contributors
2// SPDX-License-Identifier: Apache-2.0
3
4//! Opt-in forgetting and consolidation driven by policies. Nothing here runs unless called;
5//! the default policies plan nothing.
6
7use anyhow::Result;
8
9use super::recall::evaluate;
10use super::stamp::{stable_hash, Stamp};
11use super::types::*;
12use super::{ConsolidationPolicy, FactSummary, ForgettingPolicy, Memory, MemoryError};
13
14impl Memory {
15    /// Every stored fact as a maintenance policy sees it, ordered by id. Plain documents
16    /// (without memory fields) are not facts and are left out.
17    /// Panics when concurrent writes prevent a coherent snapshot; use [`Self::try_summaries`]
18    /// to handle that retryable failure.
19    pub fn summaries(&self, with_text: bool) -> Vec<FactSummary> {
20        self.try_summaries(with_text)
21            .expect("coherent memory summaries")
22    }
23
24    pub fn try_summaries(&self, with_text: bool) -> Result<Vec<FactSummary>> {
25        let now = self.clock.now_millis();
26        self.summaries_snapshot(with_text, now)
27            .map(|(_, facts)| facts)
28    }
29
30    fn summaries_snapshot(&self, with_text: bool, now: Millis) -> Result<(u64, Vec<FactSummary>)> {
31        self.with_snapshot(|index| {
32            let mut facts = Vec::new();
33            let revision = self.core.scan_documents(|document| {
34                let Some(stamp) = Stamp::from_parts(document.version, document.metadata) else {
35                    return;
36                };
37                let edges = index.edges(document.id);
38                let status =
39                    evaluate(Some(&stamp), edges, now, None).unwrap_or(FactStatus::Current);
40                facts.push(FactSummary {
41                    id: document.id.to_owned(),
42                    text: if with_text { document.text() } else { None },
43                    status,
44                    ingest_time: Some(stamp.ingest),
45                    event_time: stamp.event,
46                    valid_from: stamp.valid_from,
47                    valid_to: stamp.valid_to,
48                    effective_from: stamp.effective_from(),
49                    subject: stamp.subject.clone(),
50                    superseded_at: edges.iter().map(|edge| edge.effective).min(),
51                });
52            });
53            facts.sort_by(|a, b| a.id.cmp(&b.id));
54            check!(
55                index.revision == Some(revision),
56                "snapshot revision changed"
57            );
58            Ok((revision, facts))
59        })
60    }
61
62    /// Forgets what the configured forgetting policy selects. With the default policy that
63    /// is nothing.
64    pub fn apply_forgetting(&self, options: &ForgetOptions) -> Result<ForgetReport> {
65        self.apply_forgetting_with(self.forgetting.as_ref(), options)
66    }
67
68    pub fn apply_forgetting_with(
69        &self,
70        policy: &dyn ForgettingPolicy,
71        options: &ForgetOptions,
72    ) -> Result<ForgetReport> {
73        let now = self.clock.now_millis();
74        let mut completed = std::collections::BTreeSet::new();
75        for _ in 0..16 {
76            let (revision, facts) = self.summaries_snapshot(policy.needs_text(), now)?;
77            let known: std::collections::HashSet<&str> =
78                facts.iter().map(|fact| fact.id.as_str()).collect();
79            let mut ids: Vec<String> = policy
80                .select(&facts, now)
81                .into_iter()
82                .filter(|id| known.contains(id.as_str()))
83                .collect();
84            ids.sort_unstable();
85            ids.dedup();
86            if let Some(report) = self.forget_at_revision(
87                &ForgetScope::Ids { ids },
88                options,
89                revision,
90                &mut completed,
91            )? {
92                return Ok(report);
93            }
94        }
95        Err(MemoryError::Concurrent(
96            "memory changed during forgetting policy evaluation; retry the operation".into(),
97        )
98        .into())
99    }
100
101    /// Runs the configured consolidation policy. With the default policy that is nothing.
102    pub fn consolidate(&self, options: &ConsolidateOptions) -> Result<ConsolidationReport> {
103        self.consolidate_with(self.consolidation.as_ref(), options)
104    }
105
106    /// Writes each planned summary as a fact that supersedes its sources, so the sources stay
107    /// as history. Sources are removed only when the action asks and `allow_destructive` is
108    /// set. Summary ids derive from their sources, so repeating a run adds nothing.
109    pub fn consolidate_with(
110        &self,
111        policy: &dyn ConsolidationPolicy,
112        options: &ConsolidateOptions,
113    ) -> Result<ConsolidationReport> {
114        let now = self.clock.now_millis();
115        for _ in 0..16 {
116            let (revision, facts) = self.summaries_snapshot(policy.needs_text(), now)?;
117            let mut inputs = Vec::new();
118            let mut doomed = Vec::new();
119            for action in policy.plan(&facts, now) {
120                if action.sources.is_empty() {
121                    continue;
122                }
123                let mut sources = action.sources.clone();
124                sources.sort_unstable();
125                sources.dedup();
126                let mut summary = action.summary;
127                summary.supersedes = Some(sources.clone());
128                if summary.id.is_none() {
129                    let parts: Vec<&[u8]> = sources.iter().map(|s| s.as_bytes()).collect();
130                    summary.id = Some(format!("mem:merged:{}", stable_hash(&parts)));
131                }
132                if action.forget_sources && options.allow_destructive {
133                    doomed.extend(sources);
134                }
135                inputs.push(summary);
136            }
137            if inputs.is_empty() {
138                if !self.core.memory_revision_matches(revision) {
139                    continue;
140                }
141                return Ok(ConsolidationReport {
142                    created: Vec::new(),
143                    forgotten: Vec::new(),
144                    dry_run: options.dry_run,
145                });
146            }
147            if options.dry_run {
148                if !self.core.memory_revision_matches(revision) {
149                    continue;
150                }
151                let now_for_plan = now;
152                let created = inputs
153                    .iter()
154                    .map(|input| Remembered {
155                        id: input.id.clone().unwrap_or_default(),
156                        created: true,
157                        supersedes: input.supersedes.clone().unwrap_or_default(),
158                        superseded_by: Vec::new(),
159                        ingest_time: now_for_plan,
160                    })
161                    .collect();
162                doomed.sort_unstable();
163                doomed.dedup();
164                return Ok(ConsolidationReport {
165                    created,
166                    forgotten: doomed,
167                    dry_run: true,
168                });
169            }
170            let Some(created) = self.remember_at_revision(inputs, now, revision)? else {
171                continue;
172            };
173            let forgotten = if doomed.is_empty() {
174                Vec::new()
175            } else {
176                let expected =
177                    revision.wrapping_add(u64::from(created.iter().any(|fact| fact.created)));
178                let cleanup = self.forget_at_revision(
179                &ForgetScope::Ids { ids: doomed },
180                &ForgetOptions::default(),
181                expected,
182                &mut Default::default(),
183            ).and_then(|report| report.ok_or_else(|| MemoryError::Concurrent(
184                "consolidation committed; sources changed before cleanup; retained sources require reconciliation".into()
185            ).into()));
186                crate::core::error::after_commit("consolidation source cleanup", cleanup)?.forgotten
187            };
188            return Ok(ConsolidationReport {
189                created,
190                forgotten,
191                dry_run: false,
192            });
193        }
194        Err(MemoryError::Concurrent(
195            "memory changed during consolidation policy evaluation; retry the operation".into(),
196        )
197        .into())
198    }
199}
writerslogic_warm/core/memory_api/maintenance.rs

writerslogic_warm/core/memory_api/
maintenance.rs
