mnemo_core/query/
remember.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4use crate::error::{Error, Result};
5use crate::hash::{compute_chain_hash, compute_content_hash};
6use crate::model::capability::Capability;
7use crate::model::event::{AgentEvent, EventType};
8use crate::model::memory::{ConsolidationState, MemoryRecord, MemoryType, Scope, SourceType};
9use crate::model::relation::Relation;
10use crate::model::write_provenance::{WriteFlag, WriteOp};
11use crate::opaque_reasoning;
12use crate::query::MnemoEngine;
13#[allow(unused_imports)]
14use base64::Engine as _;
15
16#[derive(Debug, Clone, Serialize, Deserialize)]
17pub struct RememberRequest {
18    pub content: String,
19    pub agent_id: Option<String>,
20    pub memory_type: Option<MemoryType>,
21    pub scope: Option<Scope>,
22    pub importance: Option<f32>,
23    pub tags: Option<Vec<String>>,
24    pub metadata: Option<serde_json::Value>,
25    pub source_type: Option<SourceType>,
26    pub source_id: Option<String>,
27    pub org_id: Option<String>,
28    pub thread_id: Option<String>,
29    pub ttl_seconds: Option<u64>,
30    pub related_to: Option<Vec<String>>,
31    pub decay_rate: Option<f32>,
32    pub created_by: Option<String>,
33}
34
35impl RememberRequest {
36    pub fn new(content: String) -> Self {
37        Self {
38            content,
39            agent_id: None,
40            memory_type: None,
41            scope: None,
42            importance: None,
43            tags: None,
44            metadata: None,
45            source_type: None,
46            source_id: None,
47            org_id: None,
48            thread_id: None,
49            ttl_seconds: None,
50            related_to: None,
51            decay_rate: None,
52            created_by: None,
53        }
54    }
55}
56
57#[non_exhaustive]
58#[derive(Debug, Clone, Serialize, Deserialize)]
59pub struct RememberResponse {
60    pub id: Uuid,
61    pub content_hash: String,
62}
63
64impl RememberResponse {
65    pub fn new(id: Uuid, content_hash: String) -> Self {
66        Self { id, content_hash }
67    }
68}
69
70pub async fn execute(engine: &MnemoEngine, request: RememberRequest) -> Result<RememberResponse> {
71    remember_inner(engine, request, None).await
72}
73
74/// REMEMBER authorised by a verifiable [`Capability`]. The capability is verified
75/// against the engine's issuer before the write, and its id is recorded in the
76/// write provenance (the principal comes from the capability, not `created_by`).
77pub async fn execute_with_capability(
78    engine: &MnemoEngine,
79    request: RememberRequest,
80    capability: &Capability,
81) -> Result<RememberResponse> {
82    engine.verify_capability(capability)?;
83    remember_inner(engine, request, Some(capability)).await
84}
85
86async fn remember_inner(
87    engine: &MnemoEngine,
88    request: RememberRequest,
89    capability: Option<&Capability>,
90) -> Result<RememberResponse> {
91    // Validate
92    if request.content.trim().is_empty() {
93        return Err(Error::Validation("content cannot be empty".to_string()));
94    }
95
96    let resolved_tier = request.memory_type.unwrap_or(MemoryType::Episodic);
97
98    // Tier-specific importance enforcement:
99    // Procedural memories (system prompts, tool definitions) carry an
100    // importance floor so they never decay below the recall threshold.
101    let mut importance = request.importance.unwrap_or(0.5);
102    if resolved_tier == MemoryType::Procedural && importance < engine.procedural_importance_floor {
103        importance = engine.procedural_importance_floor;
104    }
105    if !(0.0..=1.0).contains(&importance) {
106        return Err(Error::Validation(
107            "importance must be between 0.0 and 1.0".to_string(),
108        ));
109    }
110
111    let agent_id = request
112        .agent_id
113        .unwrap_or_else(|| engine.default_agent_id.clone());
114    super::validate_agent_id(&agent_id)?;
115    let org_id = request.org_id.or_else(|| engine.default_org_id.clone());
116    let now = chrono::Utc::now();
117    let now_str = now.to_rfc3339();
118    let id = Uuid::now_v7();
119
120    // Compute embedding
121    let embedding = engine.embedding.embed(&request.content).await?;
122
123    // Compute content hash
124    let content_hash = compute_content_hash(&request.content, &agent_id, &now_str);
125
126    // Chain linking: look up prev_hash
127    // NOTE: Concurrent writes for the same agent_id may race on prev_hash lookup.
128    // DuckDB mode serializes via Arc<Mutex<Connection>>. PostgreSQL deployments
129    // should rely on verify_chain() to detect any broken links.
130    let prev_hash_raw = engine
131        .storage
132        .get_latest_memory_hash(&agent_id, request.thread_id.as_deref())
133        .await?;
134    let prev_hash = Some(compute_chain_hash(&content_hash, prev_hash_raw.as_deref()));
135
136    // Compute expires_at from ttl_seconds. Working-tier memories get an
137    // automatic TTL so they can't outlive their session — caller-supplied
138    // ttl_seconds still wins.
139    let effective_ttl = request.ttl_seconds.or_else(|| {
140        if resolved_tier == MemoryType::Working {
141            Some(engine.ttl_working_seconds)
142        } else {
143            None
144        }
145    });
146    let expires_at =
147        effective_ttl.map(|ttl| (now + chrono::Duration::seconds(ttl as i64)).to_rfc3339());
148
149    let mut record = MemoryRecord {
150        id,
151        agent_id: agent_id.clone(),
152        content: request.content,
153        memory_type: resolved_tier,
154        scope: request.scope.unwrap_or(Scope::Private),
155        importance,
156        tags: request.tags.unwrap_or_default(),
157        metadata: request
158            .metadata
159            .unwrap_or(serde_json::Value::Object(serde_json::Map::new())),
160        embedding: Some(embedding.clone()),
161        content_hash: content_hash.clone(),
162        prev_hash,
163        source_type: request.source_type.unwrap_or(SourceType::Agent),
164        source_id: request.source_id,
165        consolidation_state: ConsolidationState::Raw,
166        access_count: 0,
167        org_id,
168        thread_id: request.thread_id,
169        created_at: now_str.clone(),
170        updated_at: now_str,
171        last_accessed_at: None,
172        expires_at,
173        deleted_at: None,
174        decay_rate: request.decay_rate,
175        created_by: request.created_by,
176        version: 1,
177        prev_version_id: None,
178        quarantined: false,
179        quarantine_reason: None,
180        decay_function: None,
181    };
182
183    // Opaque-reasoning-payload SHAPE check (arXiv:2608.09867). Run on the
184    // PLAINTEXT content, BEFORE any encryption below — at-rest encryption
185    // base64-encodes the content and would itself look like an opaque blob. We
186    // flag the shape and record it on provenance; we do NOT reject the write
187    // (warn-and-record) and we NEVER decode the content. A flag is not proof of a
188    // secret — see crate::opaque_reasoning.
189    let mut write_flags: Vec<WriteFlag> = Vec::new();
190    if let Some(reason) = opaque_reasoning::detect(&record.content) {
191        tracing::warn!(
192            memory_id = %record.id,
193            reason = reason,
194            "remembered content has the shape of a provider opaque reasoning payload \
195             (arXiv:2608.09867); recording an opaque_reasoning_payload flag on its \
196             provenance. Shape only — this is NOT proof a secret is present. Revoke via \
197             forget_by_principal / forget_by_session if needed."
198        );
199        write_flags.push(WriteFlag::OpaqueReasoningPayload);
200    }
201
202    // Encrypt content if encryption is configured (after embedding, before storage)
203    if let Some(ref enc) = engine.encryption {
204        let encrypted = enc.encrypt(record.content.as_bytes())?;
205        record.content =
206            base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &encrypted);
207    }
208
209    // Store in database
210    engine.storage.insert_memory(&record).await?;
211
212    // Write provenance: who wrote this, under what authority. The principal is
213    // the capability holder if the write was capability-authorised, else the
214    // record's `created_by`, else its `agent_id`. Session/trace is the
215    // `thread_id`. Chained + tamper-evident (see model::write_provenance).
216    let principal = capability
217        .map(|c| c.principal.clone())
218        .or_else(|| record.created_by.clone())
219        .unwrap_or_else(|| record.agent_id.clone());
220    engine
221        .record_write_provenance(
222            record.id,
223            principal,
224            capability.map(|c| c.id),
225            record.thread_id.clone(),
226            WriteOp::Remember,
227            write_flags,
228        )
229        .await?;
230
231    // Add to vector index
232    engine.index.add(id, &embedding)?;
233
234    // Add to full-text index if available
235    if let Some(ref ft) = engine.full_text {
236        ft.add(id, &record.content)?;
237        ft.commit()?;
238    }
239
240    // Check for anomaly and update agent profile
241    let anomaly_result = super::poisoning::check_for_anomaly(engine, &record).await?;
242    if anomaly_result.is_anomalous {
243        super::poisoning::quarantine_memory(engine, id, &anomaly_result.reasons.join("; ")).await?;
244        tracing::warn!(
245            memory_id = %id,
246            score = anomaly_result.score,
247            reasons = ?anomaly_result.reasons,
248            "Memory quarantined due to anomaly detection"
249        );
250    }
251    super::poisoning::update_agent_profile(engine, &record).await?;
252
253    // Create relations if specified
254    if let Some(ref related_ids) = request.related_to {
255        for target_str in related_ids {
256            if let Ok(target_id) = Uuid::parse_str(target_str) {
257                let relation = Relation {
258                    id: Uuid::now_v7(),
259                    source_id: id,
260                    target_id,
261                    relation_type: "related_to".to_string(),
262                    weight: 1.0,
263                    metadata: serde_json::Value::Object(serde_json::Map::new()),
264                    created_at: record.created_at.clone(),
265                };
266                if let Err(e) = engine.storage.insert_relation(&relation).await {
267                    tracing::error!(relation_id = %relation.id, error = %e, "failed to insert relation");
268                }
269            }
270        }
271    }
272
273    // Emit MemoryWrite event with hash chain linking (fire-and-forget)
274    let prev_event_hash = match engine
275        .storage
276        .get_latest_event_hash(&agent_id, record.thread_id.as_deref())
277        .await
278    {
279        Ok(hash) => hash,
280        Err(e) => {
281            tracing::warn!(error = %e, "failed to get latest event hash, starting new chain segment");
282            None
283        }
284    };
285    let event_prev_hash = Some(compute_chain_hash(
286        &content_hash,
287        prev_event_hash.as_deref(),
288    ));
289    let mut event = AgentEvent {
290        id: Uuid::now_v7(),
291        agent_id: record.agent_id.clone(),
292        thread_id: record.thread_id.clone(),
293        run_id: None,
294        parent_event_id: None,
295        event_type: EventType::MemoryWrite,
296        payload: serde_json::json!({"memory_id": id.to_string()}),
297        trace_id: None,
298        span_id: None,
299        model: None,
300        tokens_input: None,
301        tokens_output: None,
302        latency_ms: None,
303        cost_usd: None,
304        timestamp: record.created_at.clone(),
305        logical_clock: 0,
306        content_hash: content_hash.clone(),
307        prev_hash: event_prev_hash,
308        embedding: None,
309    };
310    // Optionally embed the event payload
311    if engine.embed_events
312        && let Ok(emb) = engine.embedding.embed(&event.payload.to_string()).await
313    {
314        event.embedding = Some(emb);
315    }
316    if let Err(e) = engine.storage.insert_event(&event).await {
317        tracing::error!(event_id = %event.id, error = %e, "failed to insert audit event");
318    }
319
320    // Put in cache if configured
321    if let Some(ref cache) = engine.cache {
322        cache.put(record);
323    }
324
325    let hash_hex = hex::encode(&content_hash);
326
327    Ok(RememberResponse {
328        id,
329        content_hash: hash_hex,
330    })
331}
mnemo_core/query/remember.rs

mnemo_core/query/
remember.rs
