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 let (owner_scope, owner_key) = owner_for_scope(project, scope);
339 let target_project = (owner_scope == "repo").then_some(project);
340 if let Some(topic_key) = topic_key.filter(|topic_key| !topic_key.is_empty()) {
341 let existing = conn
342 .query_row(
343 "SELECT id, title, content, status, files, branch, expires_at_epoch FROM memories
344 WHERE (?3 = 'global' OR project = ?1) AND topic_key = ?2 AND scope = ?3
345 AND memory_type = ?4 AND branch IS ?5
346 AND COALESCE(owner_scope,
347 CASE WHEN scope = 'global' THEN 'user' ELSE 'repo' END) = ?6
348 AND COALESCE(owner_key,
349 CASE WHEN scope = 'global' THEN 'user:default' ELSE project END) = ?7
350 AND CASE
351 WHEN COALESCE(owner_scope,
352 CASE WHEN scope = 'global' THEN 'user' ELSE 'repo' END) = 'repo'
353 THEN COALESCE(target_project, project)
354 ELSE target_project
355 END IS ?8
356 ORDER BY CASE status WHEN 'active' THEN 0 ELSE 1 END,
357 updated_at_epoch DESC,
358 id DESC
359 LIMIT 1",
360 params![
361 project,
362 topic_key,
363 scope,
364 memory_type,
365 branch,
366 owner_scope,
367 owner_key,
368 target_project,
369 ],
370 map_existing_memory,
371 )
372 .optional()
373 .context("check existing durable memory for topic_key upsert")?;
374 if let Some(memory) = existing {
375 return Ok(Some(ExistingMemoryMatch {
376 memory,
377 source: ExistingMemoryMatchSource::TopicKey,
378 }));
379 }
380 }
381
382 if let Some(state_key) = state_key {
383 if let Some(id) = crate::memory::state_key::current_memory_id(
384 conn,
385 owner_scope,
386 &owner_key,
387 memory_type,
388 state_key,
389 now_epoch,
390 )? {
391 if let Some(memory) = load_existing_memory(conn, id)? {
392 if memory.branch.as_deref() == branch {
393 return Ok(Some(ExistingMemoryMatch {
394 memory,
395 source: ExistingMemoryMatchSource::StateKey,
396 }));
397 }
398 }
399 }
400 }
401
402 if memory_type == "preference" {
403 let (owner_scope, owner_key) = owner_for_scope(project, scope);
404 if let Some(preference_match) =
405 crate::memory::preference::consolidation::find_preference_consolidation(
406 conn,
407 owner_scope,
408 &owner_key,
409 scope,
410 branch,
411 content,
412 now_epoch,
413 )?
414 {
415 if let Some(memory) = load_existing_memory(conn, preference_match.memory_id)? {
416 return Ok(Some(ExistingMemoryMatch {
417 memory,
418 source: ExistingMemoryMatchSource::PreferenceConsolidation {
419 kind: preference_match.kind,
420 reason: preference_match.reason,
421 },
422 }));
423 }
424 }
425 }
426
427 if let Some(duplicate) = crate::memory::semantic_dedup::find_curated_duplicate(
428 conn,
429 project,
430 scope,
431 memory_type,
432 title,
433 content,
434 topic_key,
435 branch,
436 now_epoch,
437 )? {
438 if let Some(memory) = load_existing_memory(conn, duplicate.memory_id)? {
439 return Ok(Some(ExistingMemoryMatch {
440 memory,
441 source: ExistingMemoryMatchSource::Semantic {
442 similarity: duplicate.similarity,
443 },
444 }));
445 }
446 }
447
448 Ok(None)
449}
450
451fn load_existing_memory(conn: &Connection, id: i64) -> Result<Option<ExistingMemory>> {
452 conn.query_row(
453 "SELECT id, title, content, status, files, branch, expires_at_epoch
454 FROM memories WHERE id = ?1",
455 [id],
456 map_existing_memory,
457 )
458 .optional()
459 .with_context(|| format!("load existing memory for state key id={id}"))
460}
461
462fn map_existing_memory(row: &rusqlite::Row<'_>) -> rusqlite::Result<ExistingMemory> {
463 Ok(ExistingMemory {
464 id: row.get(0)?,
465 title: row.get(1)?,
466 content: row.get(2)?,
467 status: row.get(3)?,
468 files: row.get(4)?,
469 branch: row.get(5)?,
470 expires_at_epoch: row.get(6)?,
471 })
472}
473
474pub fn same_memory_text(left: &str, right: &str) -> bool {
475 normalize_memory_text(left) == normalize_memory_text(right)
476}
477
478fn normalize_memory_text(value: &str) -> String {
479 value
480 .split_whitespace()
481 .collect::<Vec<_>>()
482 .join(" ")
483 .to_ascii_lowercase()
484}