mnemo_core/query/
forget.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4use crate::error::{Error, Result};
5use crate::hash::compute_content_hash;
6use crate::model::acl::Permission;
7use crate::model::event::{AgentEvent, EventType};
8use crate::model::memory::MemoryType;
9use crate::query::MnemoEngine;
10use crate::storage::MemoryFilter;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
13#[serde(rename_all = "snake_case")]
14pub enum ForgetStrategy {
15    SoftDelete,
16    HardDelete,
17    Decay,
18    Consolidate,
19    Archive,
20    /// GDPR-aligned content erasure: replace the content with a redaction
21    /// marker while keeping the existing `content_hash` and `prev_hash`
22    /// intact so the audit trail (who wrote when) remains verifiable.
23    Redact,
24}
25
26/// Sentinel content written in place of redacted memories.
27pub const REDACTED_CONTENT: &str = "[REDACTED]";
28
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct ForgetCriteria {
31    pub max_age_hours: Option<f64>,
32    pub min_importance_below: Option<f32>,
33    pub memory_type: Option<MemoryType>,
34    pub tags: Option<Vec<String>>,
35}
36
37#[derive(Debug, Clone, Serialize, Deserialize)]
38pub struct ForgetRequest {
39    pub memory_ids: Vec<Uuid>,
40    pub agent_id: Option<String>,
41    pub strategy: Option<ForgetStrategy>,
42    pub criteria: Option<ForgetCriteria>,
43}
44
45impl ForgetRequest {
46    pub fn new(memory_ids: Vec<Uuid>) -> Self {
47        Self {
48            memory_ids,
49            agent_id: None,
50            strategy: None,
51            criteria: None,
52        }
53    }
54}
55
56#[non_exhaustive]
57#[derive(Debug, Clone, Serialize, Deserialize)]
58pub struct ForgetResponse {
59    pub forgotten: Vec<Uuid>,
60    pub errors: Vec<ForgetError>,
61}
62
63impl ForgetResponse {
64    pub fn new(forgotten: Vec<Uuid>, errors: Vec<ForgetError>) -> Self {
65        Self { forgotten, errors }
66    }
67}
68
69#[derive(Debug, Clone, Serialize, Deserialize)]
70pub struct ForgetError {
71    pub id: Uuid,
72    pub error: String,
73}
74
75pub async fn execute(engine: &MnemoEngine, request: ForgetRequest) -> Result<ForgetResponse> {
76    let agent_id = request
77        .agent_id
78        .unwrap_or_else(|| engine.default_agent_id.clone());
79    let strategy = request.strategy.unwrap_or(ForgetStrategy::SoftDelete);
80
81    // If criteria is specified and memory_ids is empty, find matching memories
82    let memory_ids = if request.memory_ids.is_empty() {
83        if let Some(ref criteria) = request.criteria {
84            let filter = MemoryFilter {
85                agent_id: Some(agent_id.clone()),
86                memory_type: criteria.memory_type,
87                min_importance: None, // We'll filter below
88                tags: criteria.tags.clone(),
89                include_deleted: false,
90                ..Default::default()
91            };
92            let memories = engine.storage.list_memories(&filter, 1000, 0).await?;
93            let now = chrono::Utc::now();
94            memories
95                .into_iter()
96                .filter(|m| {
97                    if let Some(max_age) = criteria.max_age_hours
98                        && let Ok(created) = chrono::DateTime::parse_from_rfc3339(&m.created_at)
99                    {
100                        let age_hours = (now - created.with_timezone(&chrono::Utc)).num_seconds()
101                            as f64
102                            / 3600.0;
103                        if age_hours < max_age {
104                            return false;
105                        }
106                    }
107                    if let Some(min_below) = criteria.min_importance_below
108                        && m.importance >= min_below
109                    {
110                        return false;
111                    }
112                    true
113                })
114                .map(|m| m.id)
115                .collect()
116        } else {
117            return Err(Error::Validation(
118                "memory_ids or criteria must be provided".to_string(),
119            ));
120        }
121    } else {
122        request.memory_ids.clone()
123    };
124
125    if memory_ids.is_empty() {
126        return Ok(ForgetResponse {
127            forgotten: vec![],
128            errors: vec![],
129        });
130    }
131
132    let mut forgotten = Vec::new();
133    let mut errors = Vec::new();
134
135    for id in &memory_ids {
136        // Check permission
137        match engine
138            .storage
139            .check_permission(*id, &agent_id, Permission::Write)
140            .await
141        {
142            Ok(true) => {}
143            Ok(false) => {
144                errors.push(ForgetError {
145                    id: *id,
146                    error: "permission denied".to_string(),
147                });
148                continue;
149            }
150            Err(e) => {
151                errors.push(ForgetError {
152                    id: *id,
153                    error: e.to_string(),
154                });
155                continue;
156            }
157        }
158
159        // Execute strategy
160        match strategy {
161            ForgetStrategy::SoftDelete => match engine.storage.soft_delete_memory(*id).await {
162                Ok(()) => {
163                    if let Err(e) = engine.index.remove(*id) {
164                        tracing::error!(memory_id = %id, error = %e, "failed to remove from vector index during soft delete");
165                    }
166                    if let Some(ref ft) = engine.full_text {
167                        if let Err(e) = ft.remove(*id) {
168                            tracing::error!(memory_id = %id, error = %e, "failed to remove from full-text index");
169                        }
170                        if let Err(e) = ft.commit() {
171                            tracing::error!(memory_id = %id, error = %e, "failed to commit full-text index");
172                        }
173                    }
174                    forgotten.push(*id);
175                }
176                Err(e) => {
177                    errors.push(ForgetError {
178                        id: *id,
179                        error: e.to_string(),
180                    });
181                }
182            },
183            ForgetStrategy::HardDelete => match engine.storage.hard_delete_memory(*id).await {
184                Ok(()) => {
185                    if let Err(e) = engine.index.remove(*id) {
186                        tracing::error!(memory_id = %id, error = %e, "failed to remove from vector index during hard delete");
187                    }
188                    if let Some(ref ft) = engine.full_text {
189                        if let Err(e) = ft.remove(*id) {
190                            tracing::error!(memory_id = %id, error = %e, "failed to remove from full-text index");
191                        }
192                        if let Err(e) = ft.commit() {
193                            tracing::error!(memory_id = %id, error = %e, "failed to commit full-text index");
194                        }
195                    }
196                    forgotten.push(*id);
197                }
198                Err(e) => {
199                    errors.push(ForgetError {
200                        id: *id,
201                        error: e.to_string(),
202                    });
203                }
204            },
205            ForgetStrategy::Decay => match engine.storage.get_memory(*id).await {
206                Ok(Some(mut record)) => {
207                    let decay_rate = record.decay_rate.unwrap_or(0.1);
208                    record.importance = (record.importance - decay_rate).max(0.0);
209                    record.updated_at = chrono::Utc::now().to_rfc3339();
210                    match engine.storage.update_memory(&record).await {
211                        Ok(()) => forgotten.push(*id),
212                        Err(e) => errors.push(ForgetError {
213                            id: *id,
214                            error: e.to_string(),
215                        }),
216                    }
217                }
218                Ok(None) => errors.push(ForgetError {
219                    id: *id,
220                    error: "not found".to_string(),
221                }),
222                Err(e) => errors.push(ForgetError {
223                    id: *id,
224                    error: e.to_string(),
225                }),
226            },
227            ForgetStrategy::Archive => {
228                match engine.storage.get_memory(*id).await {
229                    Ok(Some(mut record)) => {
230                        record.consolidation_state =
231                            crate::model::memory::ConsolidationState::Archived;
232                        record.updated_at = chrono::Utc::now().to_rfc3339();
233                        match engine.storage.update_memory(&record).await {
234                            Ok(()) => {
235                                // Archive to cold storage if configured
236                                if let Some(ref cs) = engine.cold_storage
237                                    && let Err(e) = cs.archive(&record).await
238                                {
239                                    tracing::warn!("cold storage archive failed for {}: {e}", id);
240                                }
241                                forgotten.push(*id);
242                            }
243                            Err(e) => errors.push(ForgetError {
244                                id: *id,
245                                error: e.to_string(),
246                            }),
247                        }
248                    }
249                    Ok(None) => errors.push(ForgetError {
250                        id: *id,
251                        error: "not found".to_string(),
252                    }),
253                    Err(e) => errors.push(ForgetError {
254                        id: *id,
255                        error: e.to_string(),
256                    }),
257                }
258            }
259            ForgetStrategy::Consolidate => match engine.storage.get_memory(*id).await {
260                Ok(Some(mut record)) => {
261                    record.consolidation_state =
262                        crate::model::memory::ConsolidationState::Consolidated;
263                    record.updated_at = chrono::Utc::now().to_rfc3339();
264                    match engine.storage.update_memory(&record).await {
265                        Ok(()) => forgotten.push(*id),
266                        Err(e) => errors.push(ForgetError {
267                            id: *id,
268                            error: e.to_string(),
269                        }),
270                    }
271                }
272                Ok(None) => errors.push(ForgetError {
273                    id: *id,
274                    error: "not found".to_string(),
275                }),
276                Err(e) => errors.push(ForgetError {
277                    id: *id,
278                    error: e.to_string(),
279                }),
280            },
281            ForgetStrategy::Redact => {
282                // GDPR erasure: replace content, leave content_hash + prev_hash
283                // untouched so downstream chain verification still works.
284                match engine.storage.get_memory(*id).await {
285                    Ok(Some(mut record)) => {
286                        record.content = REDACTED_CONTENT.to_string();
287                        record.tags.retain(|t| !t.starts_with("subject:"));
288                        record.metadata = serde_json::json!({"redacted": true});
289                        record.updated_at = chrono::Utc::now().to_rfc3339();
290                        match engine.storage.update_memory(&record).await {
291                            Ok(()) => {
292                                if let Err(e) = engine.index.remove(*id) {
293                                    tracing::error!(memory_id = %id, error = %e, "failed to remove from vector index during redact");
294                                }
295                                if let Some(ref ft) = engine.full_text {
296                                    if let Err(e) = ft.remove(*id) {
297                                        tracing::error!(memory_id = %id, error = %e, "failed to remove from full-text index during redact");
298                                    }
299                                    if let Err(e) = ft.commit() {
300                                        tracing::error!(memory_id = %id, error = %e, "failed to commit full-text index during redact");
301                                    }
302                                }
303                                if let Some(ref cache) = engine.cache {
304                                    cache.invalidate(*id);
305                                }
306                                forgotten.push(*id);
307                            }
308                            Err(e) => errors.push(ForgetError {
309                                id: *id,
310                                error: e.to_string(),
311                            }),
312                        }
313                    }
314                    Ok(None) => errors.push(ForgetError {
315                        id: *id,
316                        error: "not found".to_string(),
317                    }),
318                    Err(e) => errors.push(ForgetError {
319                        id: *id,
320                        error: e.to_string(),
321                    }),
322                }
323            }
324        }
325    }
326
327    // Emit MemoryDelete event for each forgotten memory with hash chaining (fire-and-forget)
328    let now = chrono::Utc::now().to_rfc3339();
329    for id in &forgotten {
330        let event_content_hash = compute_content_hash(&id.to_string(), &agent_id, &now);
331        let prev_event_hash = match engine.storage.get_latest_event_hash(&agent_id, None).await {
332            Ok(hash) => hash,
333            Err(e) => {
334                tracing::warn!(error = %e, "failed to get latest event hash, starting new chain segment");
335                None
336            }
337        };
338        let event_prev_hash = Some(crate::hash::compute_chain_hash(
339            &event_content_hash,
340            prev_event_hash.as_deref(),
341        ));
342        let event = AgentEvent {
343            id: Uuid::now_v7(),
344            agent_id: agent_id.clone(),
345            thread_id: None,
346            run_id: None,
347            parent_event_id: None,
348            event_type: EventType::MemoryDelete,
349            payload: serde_json::json!({"memory_id": id.to_string()}),
350            trace_id: None,
351            span_id: None,
352            model: None,
353            tokens_input: None,
354            tokens_output: None,
355            latency_ms: None,
356            cost_usd: None,
357            timestamp: now.clone(),
358            logical_clock: 0,
359            content_hash: event_content_hash,
360            prev_hash: event_prev_hash,
361            embedding: None,
362        };
363        if let Err(e) = engine.storage.insert_event(&event).await {
364            tracing::error!(event_id = %event.id, error = %e, "failed to insert audit event");
365        }
366
367        // Invalidate cache on forget
368        if let Some(ref cache) = engine.cache {
369            cache.invalidate(*id);
370        }
371    }
372
373    // v0.4.10 — opportunistic, feedback-driven consolidation trigger.
374    // Best-effort: failures are logged, never propagated to the caller.
375    // Skipped under the default FixedSize policy and when the policy
376    // explicitly disables the forget hook.
377    if !forgotten.is_empty()
378        && let crate::query::maturity::ConsolidationPolicy::MaturityDriven(ref p) =
379            engine.consolidation_policy
380        && p.trigger_on_forget
381        && let Err(e) =
382            super::lifecycle::run_consolidation(engine, &agent_id, p.min_cluster_size_floor.max(1))
383                .await
384    {
385        tracing::warn!(error = %e, "post-forget maturity-driven consolidation failed (best-effort)");
386    }
387
388    Ok(ForgetResponse { forgotten, errors })
389}
390
391/// Tag convention used by :fn:`forget_subject` to locate memories owned by
392/// a given subject (e.g. end-user or data principal under GDPR/DPDPA).
393pub const SUBJECT_TAG_PREFIX: &str = "subject:";
394
395#[derive(Debug, Clone, Serialize, Deserialize)]
396pub struct ForgetSubjectRequest {
397    /// Opaque identifier for the subject whose memories should be erased.
398    /// Memories are matched via tag `subject:<subject_id>`.
399    pub subject_id: String,
400    /// Optional agent scope; defaults to the engine default.
401    pub agent_id: Option<String>,
402    /// Erasure strategy. Only `Redact` and `HardDelete` are meaningful for
403    /// a subject-scoped operation; other strategies are accepted and passed
404    /// through to the standard forget pipeline.
405    pub strategy: ForgetStrategy,
406}
407
408#[non_exhaustive]
409#[derive(Debug, Clone, Serialize, Deserialize)]
410pub struct ForgetSubjectResponse {
411    pub subject_id: String,
412    pub strategy: ForgetStrategy,
413    pub matched: usize,
414    pub forgotten: Vec<Uuid>,
415    pub cascaded_events: usize,
416    pub errors: Vec<ForgetError>,
417}
418
419/// Erase every memory tagged with `subject:<subject_id>`, using the
420/// specified strategy.
421///
422/// * `Redact` preserves the existing `content_hash` and `prev_hash`
423///   so downstream chain verification still succeeds; only the content
424///   itself becomes the REDACTED marker.
425/// * `HardDelete` removes the memory row and cascades to emit a
426///   `MemoryDelete` audit event per record; no further cascade is
427///   performed against events referencing the memory id, those remain
428///   available for audit.
429pub async fn forget_subject(
430    engine: &MnemoEngine,
431    request: ForgetSubjectRequest,
432) -> Result<ForgetSubjectResponse> {
433    if request.subject_id.is_empty() {
434        return Err(Error::Validation("subject_id cannot be empty".to_string()));
435    }
436    let agent_id = request
437        .agent_id
438        .clone()
439        .unwrap_or_else(|| engine.default_agent_id.clone());
440    super::validate_agent_id(&agent_id)?;
441
442    let subject_tag = format!("{SUBJECT_TAG_PREFIX}{}", request.subject_id);
443    // The shared storage backends don't push the tag predicate into SQL yet,
444    // so we filter in Rust. This is O(n) over the agent's non-deleted memories,
445    // which is acceptable for GDPR-erasure workloads and keeps the contract
446    // consistent across DuckDB / PostgreSQL.
447    let filter = MemoryFilter {
448        agent_id: Some(agent_id.clone()),
449        include_deleted: false,
450        ..Default::default()
451    };
452    let all_records = engine
453        .storage
454        .list_memories(&filter, super::MAX_BATCH_QUERY_LIMIT, 0)
455        .await?;
456    let matched_records: Vec<_> = all_records
457        .into_iter()
458        .filter(|r| r.tags.iter().any(|t| t == &subject_tag))
459        .collect();
460    let matched = matched_records.len();
461    let ids: Vec<Uuid> = matched_records.iter().map(|r| r.id).collect();
462
463    if ids.is_empty() {
464        return Ok(ForgetSubjectResponse {
465            subject_id: request.subject_id,
466            strategy: request.strategy,
467            matched,
468            forgotten: Vec::new(),
469            cascaded_events: 0,
470            errors: Vec::new(),
471        });
472    }
473
474    // Audit events that reference these memories are intentionally preserved:
475    // removing them would break the per-agent hash chain and destroy the
476    // GDPR-aligned audit trail. Callers who need to erase event payloads
477    // containing PII should issue a targeted Redact against the relevant
478    // event rows (not yet exposed as a public API).
479    let cascaded_events: usize = 0;
480
481    // Re-use the standard forget pipeline for the actual deletion/redaction
482    // so audit-event semantics stay consistent with mnemo.forget.
483    let standard_req = ForgetRequest {
484        memory_ids: ids,
485        agent_id: Some(agent_id.clone()),
486        strategy: Some(request.strategy),
487        criteria: None,
488    };
489    let resp = execute(engine, standard_req).await?;
490
491    // For Redact, emit a MemoryRedact event per affected memory so auditors
492    // can distinguish redactions from ordinary deletes.
493    if request.strategy == ForgetStrategy::Redact {
494        let now = chrono::Utc::now().to_rfc3339();
495        for id in &resp.forgotten {
496            let content_hash = compute_content_hash(
497                &format!("redact:{id}:{}", request.subject_id),
498                &agent_id,
499                &now,
500            );
501            let prev_hash_raw = engine
502                .storage
503                .get_latest_event_hash(&agent_id, None)
504                .await
505                .ok()
506                .flatten();
507            let event_prev_hash = Some(crate::hash::compute_chain_hash(
508                &content_hash,
509                prev_hash_raw.as_deref(),
510            ));
511            let event = AgentEvent {
512                id: Uuid::now_v7(),
513                agent_id: agent_id.clone(),
514                thread_id: None,
515                run_id: None,
516                parent_event_id: None,
517                event_type: EventType::MemoryRedact,
518                payload: serde_json::json!({
519                    "memory_id": id.to_string(),
520                    "subject_id": request.subject_id,
521                }),
522                trace_id: None,
523                span_id: None,
524                model: None,
525                tokens_input: None,
526                tokens_output: None,
527                latency_ms: None,
528                cost_usd: None,
529                timestamp: now.clone(),
530                logical_clock: 0,
531                content_hash,
532                prev_hash: event_prev_hash,
533                embedding: None,
534            };
535            if let Err(e) = engine.storage.insert_event(&event).await {
536                tracing::error!(
537                    event_id = %event.id,
538                    error = %e,
539                    "failed to insert MemoryRedact event"
540                );
541            }
542        }
543    }
544
545    Ok(ForgetSubjectResponse {
546        subject_id: request.subject_id,
547        strategy: request.strategy,
548        matched,
549        forgotten: resp.forgotten,
550        cascaded_events,
551        errors: resp.errors,
552    })
553}
mnemo_core/query/forget.rs

mnemo_core/query/
forget.rs
