1use anyhow::{Context, Result};
2use rusqlite::{params, Connection, OptionalExtension};
3
4use crate::memory::lifecycle::MemoryLifecycleOp;
5use crate::memory::preference::consolidation::PreferenceConsolidationKind;
6
7pub const PLANNER_VERSION: &str = "memory-operation-planner-v1";
8
9#[derive(Debug, Clone, PartialEq)]
10pub struct MemoryOperationInput {
11 pub source: String,
12 pub actor: String,
13 pub source_project: String,
14 pub owner_scope: String,
15 pub owner_key: String,
16 pub memory_type: String,
17 pub topic_key: Option<String>,
18 pub state_key: Option<String>,
19 pub source_candidate_id: Option<i64>,
20 pub confidence: Option<f64>,
21}
22
23#[derive(Debug, Clone, PartialEq)]
24pub struct MemoryOperationPlan {
25 pub op: MemoryLifecycleOp,
26 pub state_key: Option<String>,
27 pub target_memory_id: Option<i64>,
28 pub superseded_ids: Vec<i64>,
29 pub conflicting_ids: Vec<i64>,
30 pub noop_reason: Option<String>,
31 pub defer_reason: Option<String>,
32 pub planner_version: &'static str,
33 pub reason: String,
34}
35
36impl MemoryOperationPlan {
37 pub fn new(
38 op: MemoryLifecycleOp,
39 state_key: Option<String>,
40 reason: impl Into<String>,
41 ) -> Self {
42 Self {
43 op,
44 state_key,
45 target_memory_id: None,
46 superseded_ids: Vec::new(),
47 conflicting_ids: Vec::new(),
48 noop_reason: None,
49 defer_reason: None,
50 planner_version: PLANNER_VERSION,
51 reason: reason.into(),
52 }
53 }
54
55 pub fn with_target_memory_id(mut self, memory_id: Option<i64>) -> Self {
56 self.target_memory_id = memory_id;
57 self
58 }
59
60 pub fn with_superseded_ids(mut self, superseded_ids: Vec<i64>) -> Self {
61 self.superseded_ids = superseded_ids;
62 self
63 }
64
65 pub fn with_conflicting_ids(mut self, conflicting_ids: Vec<i64>) -> Self {
66 self.conflicting_ids = conflicting_ids;
67 self
68 }
69
70 pub fn with_noop_reason(mut self, reason: impl Into<String>) -> Self {
71 self.noop_reason = Some(reason.into());
72 self
73 }
74
75 pub fn with_defer_reason(mut self, reason: impl Into<String>) -> Self {
76 self.defer_reason = Some(reason.into());
77 self
78 }
79}
80
81pub fn owner_for_scope(project: &str, scope: &str) -> (&'static str, String) {
82 if scope == "global" {
83 ("user", "user:default".to_string())
84 } else {
85 ("repo", project.to_string())
86 }
87}
88
89#[allow(clippy::too_many_arguments)]
90pub fn plan_direct_save(
91 conn: &Connection,
92 source: &str,
93 actor: &str,
94 project: &str,
95 scope: &str,
96 memory_type: &str,
97 topic_key: Option<&str>,
98 title: &str,
99 content: &str,
100 files: Option<&str>,
101 branch: Option<&str>,
102 source_candidate_id: Option<i64>,
103 confidence: Option<f64>,
104) -> Result<(MemoryOperationInput, MemoryOperationPlan)> {
105 let now = chrono::Utc::now().timestamp();
106 let (owner_scope, owner_key) = owner_for_scope(project, scope);
107 let state_key_decision =
108 crate::memory::state_key::derive_state_key(memory_type, topic_key, title, content);
109 let state_key = state_key_decision
110 .as_ref()
111 .map(|decision| decision.state_key.clone());
112 let direct_upsert_state_key = state_key_decision
113 .as_ref()
114 .filter(|decision| decision.allows_direct_upsert())
115 .map(|decision| decision.state_key.as_str());
116 let existing_match = existing_memory_for_direct_save(
117 conn,
118 project,
119 scope,
120 memory_type,
121 topic_key,
122 direct_upsert_state_key,
123 title,
124 content,
125 branch,
126 now,
127 )?;
128 let input = MemoryOperationInput {
129 source: source.to_string(),
130 actor: actor.to_string(),
131 source_project: project.to_string(),
132 owner_scope: owner_scope.to_string(),
133 owner_key,
134 memory_type: memory_type.to_string(),
135 topic_key: topic_key.map(str::to_string),
136 state_key: state_key.clone(),
137 source_candidate_id,
138 confidence,
139 };
140 let plan = match existing_match {
141 Some(existing_match)
142 if existing_match
143 .memory
144 .matches_noop_write(title, content, files, branch, now) =>
145 {
146 MemoryOperationPlan::new(
147 MemoryLifecycleOp::Noop,
148 state_key,
149 "existing active memory already represents this fact",
150 )
151 .with_target_memory_id(Some(existing_match.memory.id))
152 .with_noop_reason("already represented by active memory")
153 }
154 Some(existing_match) => {
155 let reason = match existing_match.source {
156 ExistingMemoryMatchSource::TopicKey | ExistingMemoryMatchSource::StateKey => {
157 "existing state/topic memory will be updated".to_string()
158 }
159 ExistingMemoryMatchSource::PreferenceConsolidation {
160 kind: PreferenceConsolidationKind::Contradiction,
161 reason,
162 } => {
163 return Ok((
164 input,
165 MemoryOperationPlan::new(MemoryLifecycleOp::Conflict, state_key, reason)
166 .with_conflicting_ids(vec![existing_match.memory.id]),
167 ));
168 }
169 ExistingMemoryMatchSource::PreferenceConsolidation { reason, .. } => reason,
170 ExistingMemoryMatchSource::Semantic { similarity } => {
171 format!(
172 "semantic duplicate memory will be updated (similarity {similarity:.3})"
173 )
174 }
175 };
176 MemoryOperationPlan::new(MemoryLifecycleOp::Update, state_key, reason)
177 .with_target_memory_id(Some(existing_match.memory.id))
178 }
179 None => MemoryOperationPlan::new(
180 MemoryLifecycleOp::Add,
181 state_key,
182 "no existing state/topic memory; add new durable memory",
183 ),
184 };
185 Ok((input, plan))
186}
187
188pub fn operation_for_memory_write(conn: &Connection, memory_id: i64) -> Result<MemoryLifecycleOp> {
189 let created_updated: (i64, i64) = conn
190 .query_row(
191 "SELECT created_at_epoch, updated_at_epoch FROM memories WHERE id = ?1",
192 [memory_id],
193 |row| Ok((row.get(0)?, row.get(1)?)),
194 )
195 .with_context(|| format!("load memory timestamps for operation result id={memory_id}"))?;
196 Ok(if created_updated.0 == created_updated.1 {
197 MemoryLifecycleOp::Add
198 } else {
199 MemoryLifecycleOp::Update
200 })
201}
202
203pub fn insert_operation_log(
204 conn: &Connection,
205 input: &MemoryOperationInput,
206 plan: &MemoryOperationPlan,
207 result_memory_id: Option<i64>,
208) -> Result<i64> {
209 let now = chrono::Utc::now().timestamp();
210 let superseded_ids = serde_json::to_string(&plan.superseded_ids)
211 .context("serialize memory operation superseded ids")?;
212 let conflicting_ids = serde_json::to_string(&plan.conflicting_ids)
213 .context("serialize memory operation conflicting ids")?;
214 conn.execute(
215 "INSERT INTO memory_operation_log
216 (operation, planner_version, actor, source, owner_scope, owner_key,
217 memory_type, state_key, input_topic_key, source_candidate_id, result_memory_id,
218 superseded_ids, conflicting_ids, noop_reason, defer_reason, confidence, reason,
219 created_at_epoch)
220 VALUES (?1, ?2, ?3, ?4, ?5, ?6,
221 ?7, ?8, ?9, ?10, ?11,
222 ?12, ?13, ?14, ?15, ?16, ?17,
223 ?18)",
224 params![
225 plan.op.as_str(),
226 plan.planner_version,
227 input.actor.as_str(),
228 input.source.as_str(),
229 input.owner_scope.as_str(),
230 input.owner_key.as_str(),
231 input.memory_type.as_str(),
232 plan.state_key.as_deref().or(input.state_key.as_deref()),
233 input.topic_key.as_deref(),
234 input.source_candidate_id,
235 result_memory_id.or(plan.target_memory_id),
236 superseded_ids,
237 conflicting_ids,
238 plan.noop_reason.as_deref(),
239 plan.defer_reason.as_deref(),
240 input.confidence,
241 plan.reason.as_str(),
242 now
243 ],
244 )
245 .context("insert memory operation audit log")?;
246 Ok(conn.last_insert_rowid())
247}
248
249pub fn with_operation_savepoint<T>(conn: &Connection, f: impl FnOnce() -> Result<T>) -> Result<T> {
250 conn.execute_batch("SAVEPOINT remem_memory_operation")?;
251 match f() {
252 Ok(value) => {
253 conn.execute_batch("RELEASE SAVEPOINT remem_memory_operation")?;
254 Ok(value)
255 }
256 Err(error) => {
257 let rollback = conn.execute_batch(
258 "ROLLBACK TO SAVEPOINT remem_memory_operation;
259 RELEASE SAVEPOINT remem_memory_operation;",
260 );
261 if let Err(rollback_error) = rollback {
262 return Err(error.context(format!(
263 "memory operation rollback also failed: {rollback_error}"
264 )));
265 }
266 Err(error)
267 }
268 }
269}
270
271#[derive(Debug)]
272struct ExistingMemory {
273 id: i64,
274 title: String,
275 content: String,
276 status: String,
277 files: Option<String>,
278 branch: Option<String>,
279 expires_at_epoch: Option<i64>,
280}
281
282#[derive(Debug)]
283struct ExistingMemoryMatch {
284 memory: ExistingMemory,
285 source: ExistingMemoryMatchSource,
286}
287
288#[derive(Debug)]
289enum ExistingMemoryMatchSource {
290 TopicKey,
291 StateKey,
292 PreferenceConsolidation {
293 kind: PreferenceConsolidationKind,
294 reason: String,
295 },
296 Semantic {
297 similarity: f32,
298 },
299}
300
301impl ExistingMemory {
302 fn matches_noop_write(
303 &self,
304 title: &str,
305 content: &str,
306 files: Option<&str>,
307 branch: Option<&str>,
308 now_epoch: i64,
309 ) -> bool {
310 self.status == "active"
311 && self.is_current(now_epoch)
312 && self.title == title
313 && same_memory_text(&self.content, content)
314 && self.files.as_deref() == files
315 && self.branch.as_deref() == branch
316 }
317
318 fn is_current(&self, now_epoch: i64) -> bool {
319 match self.expires_at_epoch {
320 Some(expires_at_epoch) => expires_at_epoch > now_epoch,
321 None => true,
322 }
323 }
324}
325
326fn existing_memory_for_direct_save(
327 conn: &Connection,
328 project: &str,
329 scope: &str,
330 memory_type: &str,
331 topic_key: Option<&str>,
332 state_key: Option<&str>,
333 title: &str,
334 content: &str,
335 branch: Option<&str>,
336 now_epoch: i64,
337) -> Result<Option<ExistingMemoryMatch>> {
338 if let Some(topic_key) = topic_key.filter(|topic_key| !topic_key.is_empty()) {
339 let existing = conn
340 .query_row(
341 "SELECT id, title, content, status, files, branch, expires_at_epoch FROM memories
342 WHERE project = ?1 AND topic_key = ?2 AND scope = ?3
343 AND memory_type = ?4
344 ORDER BY CASE status WHEN 'active' THEN 0 ELSE 1 END,
345 updated_at_epoch DESC,
346 id DESC
347 LIMIT 1",
348 params![project, topic_key, scope, memory_type],
349 map_existing_memory,
350 )
351 .optional()
352 .context("check existing durable memory for topic_key upsert")?;
353 if let Some(memory) = existing {
354 return Ok(Some(ExistingMemoryMatch {
355 memory,
356 source: ExistingMemoryMatchSource::TopicKey,
357 }));
358 }
359 }
360
361 if let Some(state_key) = state_key {
362 let (owner_scope, owner_key) = owner_for_scope(project, scope);
363 if let Some(id) = crate::memory::state_key::current_memory_id(
364 conn,
365 owner_scope,
366 &owner_key,
367 memory_type,
368 state_key,
369 now_epoch,
370 )? {
371 if let Some(memory) = load_existing_memory(conn, id)? {
372 return Ok(Some(ExistingMemoryMatch {
373 memory,
374 source: ExistingMemoryMatchSource::StateKey,
375 }));
376 }
377 }
378 }
379
380 if memory_type == "preference" {
381 let (owner_scope, owner_key) = owner_for_scope(project, scope);
382 if let Some(preference_match) =
383 crate::memory::preference::consolidation::find_preference_consolidation(
384 conn,
385 owner_scope,
386 &owner_key,
387 scope,
388 branch,
389 content,
390 now_epoch,
391 )?
392 {
393 if let Some(memory) = load_existing_memory(conn, preference_match.memory_id)? {
394 return Ok(Some(ExistingMemoryMatch {
395 memory,
396 source: ExistingMemoryMatchSource::PreferenceConsolidation {
397 kind: preference_match.kind,
398 reason: preference_match.reason,
399 },
400 }));
401 }
402 }
403 }
404
405 if let Some(duplicate) = crate::memory::semantic_dedup::find_curated_duplicate(
406 conn,
407 project,
408 scope,
409 memory_type,
410 title,
411 content,
412 topic_key,
413 branch,
414 now_epoch,
415 )? {
416 if let Some(memory) = load_existing_memory(conn, duplicate.memory_id)? {
417 return Ok(Some(ExistingMemoryMatch {
418 memory,
419 source: ExistingMemoryMatchSource::Semantic {
420 similarity: duplicate.similarity,
421 },
422 }));
423 }
424 }
425
426 Ok(None)
427}
428
429fn load_existing_memory(conn: &Connection, id: i64) -> Result<Option<ExistingMemory>> {
430 conn.query_row(
431 "SELECT id, title, content, status, files, branch, expires_at_epoch
432 FROM memories WHERE id = ?1",
433 [id],
434 map_existing_memory,
435 )
436 .optional()
437 .with_context(|| format!("load existing memory for state key id={id}"))
438}
439
440fn map_existing_memory(row: &rusqlite::Row<'_>) -> rusqlite::Result<ExistingMemory> {
441 Ok(ExistingMemory {
442 id: row.get(0)?,
443 title: row.get(1)?,
444 content: row.get(2)?,
445 status: row.get(3)?,
446 files: row.get(4)?,
447 branch: row.get(5)?,
448 expires_at_epoch: row.get(6)?,
449 })
450}
451
452pub fn same_memory_text(left: &str, right: &str) -> bool {
453 normalize_memory_text(left) == normalize_memory_text(right)
454}
455
456fn normalize_memory_text(value: &str) -> String {
457 value
458 .split_whitespace()
459 .collect::<Vec<_>>()
460 .join(" ")
461 .to_ascii_lowercase()
462}