weavatrix_memory/analytics/
consolidation.rs

1use super::{ConsolidationAction, ConsolidationKind, ConsolidationPlan, MemoryAnalytics};
2use crate::{EntityId, MemoryError, MemoryProjection, ProjectionClock, Result, project_graph};
3use std::collections::BTreeMap;
4
5impl MemoryAnalytics {
6    /// Produces a deterministic, non-mutating maintenance plan.
7    ///
8    /// Event history is never deleted. The caller may translate proposed
9    /// duplicate actions into explicit supersession events.
10    ///
11    /// # Errors
12    ///
13    /// Rejects a zero action limit or graph projection failures.
14    pub fn consolidation_plan(
15        projection: &MemoryProjection,
16        clock: ProjectionClock,
17        max_actions: usize,
18    ) -> Result<ConsolidationPlan> {
19        if max_actions == 0 {
20            return Err(MemoryError::InvalidValue {
21                field: "consolidation.max_actions",
22                reason: "must be greater than zero",
23            });
24        }
25        let view = projection.view(clock);
26        let graph = project_graph(&view)?;
27        let mut actions = duplicate_actions(&view.facts);
28        actions.extend(orphan_actions(&graph));
29        actions.extend(revision_actions(projection, clock));
30        actions.sort_by(|left, right| {
31            left.kind
32                .cmp(&right.kind)
33                .then_with(|| left.affected_entities.cmp(&right.affected_entities))
34                .then_with(|| left.affected_facts.cmp(&right.affected_facts))
35        });
36        actions.truncate(max_actions);
37        let projected_savings = actions
38            .iter()
39            .map(|action| match action.kind {
40                ConsolidationKind::ReviewOrphan => 0,
41                _ => action.affected_facts.len().saturating_sub(1),
42            })
43            .sum();
44        Ok(ConsolidationPlan {
45            actions,
46            projected_savings,
47            source_position: projection.last_global_position(),
48        })
49    }
50}
51
52fn duplicate_actions(facts: &[crate::MemoryFact]) -> Vec<ConsolidationAction> {
53    let mut groups = BTreeMap::<(EntityId, String, EntityId), Vec<&crate::MemoryFact>>::new();
54    for fact in facts {
55        groups
56            .entry((
57                fact.source.clone(),
58                fact.relation.clone(),
59                fact.target.clone(),
60            ))
61            .or_default()
62            .push(fact);
63    }
64    let mut actions = Vec::new();
65    for ((source, _, target), mut duplicates) in groups {
66        if duplicates.len() < 2 {
67            continue;
68        }
69        duplicates.sort_by(|left, right| {
70            right
71                .confidence
72                .cmp(&left.confidence)
73                .then_with(|| right.recorded_at.cmp(&left.recorded_at))
74                .then_with(|| left.id.cmp(&right.id))
75        });
76        actions.push(ConsolidationAction {
77            kind: ConsolidationKind::SupersedeDuplicate,
78            keep: Some(duplicates[0].id.clone()),
79            affected_facts: duplicates.iter().map(|fact| fact.id.clone()).collect(),
80            affected_entities: vec![source, target],
81            rationale: "same active source, relation, and target; keep strongest evidence"
82                .to_owned(),
83        });
84    }
85    actions
86}
87
88fn orphan_actions(graph: &weavatrix_graph::Graph) -> Vec<ConsolidationAction> {
89    graph
90        .nodes()
91        .iter()
92        .filter_map(|node| {
93            let index = graph.node_index(node.id.as_str())?;
94            let isolated = graph.in_degree(index) == Some(0) && graph.out_degree(index) == Some(0);
95            isolated.then(|| ConsolidationAction {
96                kind: ConsolidationKind::ReviewOrphan,
97                keep: None,
98                affected_facts: Vec::new(),
99                affected_entities: EntityId::new(node.id.as_str()).into_iter().collect(),
100                rationale: "entity has no active incoming or outgoing evidence".to_owned(),
101            })
102        })
103        .collect()
104}
105
106fn revision_actions(
107    projection: &MemoryProjection,
108    clock: ProjectionClock,
109) -> Vec<ConsolidationAction> {
110    let mut groups = BTreeMap::<(EntityId, String), Vec<&crate::MemoryFact>>::new();
111    for fact in projection
112        .all_facts()
113        .iter()
114        .filter(|fact| fact.recorded_at <= clock.known_at)
115    {
116        groups
117            .entry((fact.source.clone(), fact.relation.clone()))
118            .or_default()
119            .push(fact);
120    }
121    groups
122        .into_iter()
123        .filter_map(|((entity, _), mut facts)| {
124            facts.sort_by_key(|fact| (fact.recorded_at, fact.id.clone()));
125            let chain = facts
126                .iter()
127                .filter(|fact| fact.supersedes.is_some())
128                .count();
129            (chain >= 3).then(|| ConsolidationAction {
130                kind: ConsolidationKind::CompactRevisionChain,
131                keep: facts.last().map(|fact| fact.id.clone()),
132                affected_facts: facts.iter().map(|fact| fact.id.clone()).collect(),
133                affected_entities: vec![entity],
134                rationale: "retain event history but checkpoint a long revision chain".to_owned(),
135            })
136        })
137        .collect()
138}
weavatrix_memory/analytics/consolidation.rs

weavatrix_memory/analytics/
consolidation.rs
