Skip to main content

remem/memory/service/
save.rs

1use anyhow::{anyhow, Context, Result};
2use rusqlite::Connection;
3use std::path::{Path, PathBuf};
4use std::time::{SystemTime, UNIX_EPOCH};
5
6use super::local_copy::{
7    build_local_note_content, local_copy_enabled_override, resolve_local_note_path,
8    write_local_note,
9};
10use super::types::{LocalCopyResult, SaveMemoryNextStep, SaveMemoryRequest, SaveMemoryResult};
11use crate::memory::claims::{claims_enabled, insert_memory_claim, ClaimWriteRequest};
12use crate::memory::lesson::{save_lesson_with_reference_time, SaveLessonRequest};
13use crate::memory::lifecycle::MemoryLifecycleOp;
14use crate::memory::poisoning::{
15    scan_instruction_pattern, InstructionPatternMatch, DIRECT_SAVE_TRUST_CLASS,
16};
17use crate::memory::{MemoryType, MEMORY_TYPES};
18
19#[derive(Debug)]
20pub struct LocalCopyError {
21    message: String,
22}
23
24impl From<anyhow::Error> for LocalCopyError {
25    fn from(err: anyhow::Error) -> Self {
26        Self {
27            message: err.to_string(),
28        }
29    }
30}
31
32impl std::fmt::Display for LocalCopyError {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        f.write_str(&self.message)
35    }
36}
37
38impl std::error::Error for LocalCopyError {}
39
40#[derive(Debug)]
41pub struct SaveMemoryValidationError {
42    message: String,
43}
44
45impl SaveMemoryValidationError {
46    pub(crate) fn new(message: impl Into<String>) -> Self {
47        Self {
48            message: message.into(),
49        }
50    }
51}
52
53impl std::fmt::Display for SaveMemoryValidationError {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.write_str(&self.message)
56    }
57}
58
59impl std::error::Error for SaveMemoryValidationError {}
60
61pub fn save_memory(conn: &Connection, req: &SaveMemoryRequest) -> Result<SaveMemoryResult> {
62    save_memory_with_reference_time(conn, req, req.created_at_epoch)
63}
64
65pub fn save_memory_with_reference_time(
66    conn: &Connection,
67    req: &SaveMemoryRequest,
68    reference_time_epoch: Option<i64>,
69) -> Result<SaveMemoryResult> {
70    let reference_time_epoch =
71        normalize_reference_time_epoch(req.created_at_epoch, reference_time_epoch);
72    save_memory_inner(conn, req, reference_time_epoch)
73}
74
75fn normalize_reference_time_epoch(
76    created_at_epoch: Option<i64>,
77    reference_time_epoch: Option<i64>,
78) -> Option<i64> {
79    reference_time_epoch.or(created_at_epoch)
80}
81
82fn save_memory_inner(
83    conn: &Connection,
84    req: &SaveMemoryRequest,
85    reference_time_epoch: Option<i64>,
86) -> Result<SaveMemoryResult> {
87    let validated = validate_save_memory_request(req)?;
88    let project = req.project.as_deref().unwrap_or("manual");
89    let title = req.title.as_deref().unwrap_or("Memory");
90    let memory_type = validated.memory_type.as_str();
91    let files_json = req
92        .files
93        .as_ref()
94        .and_then(|files| serde_json::to_string(files).ok());
95
96    let scope = validated.scope.as_str();
97    let effective_topic_key = effective_topic_key(req, memory_type);
98    let acknowledgement =
99        direct_save_pattern_acknowledgement(title, &req.text, req.acknowledge_pattern.as_deref())?;
100
101    let mut local_copy = prepare_local_copy(project, title, req).map_err(LocalCopyError::from)?;
102    write_local_copy(&mut local_copy).map_err(LocalCopyError::from)?;
103
104    let save_result = if memory_type == "lesson" {
105        crate::memory::operation::with_operation_savepoint(conn, || {
106            let (operation_input, operation_plan) = crate::memory::operation::plan_direct_save(
107                conn,
108                "direct",
109                "save_memory",
110                project,
111                scope,
112                memory_type,
113                effective_topic_key.as_deref(),
114                title,
115                &req.text,
116                files_json.as_deref(),
117                req.branch.as_deref(),
118                None,
119                None,
120            )?;
121            let id = save_lesson_with_reference_time(
122                conn,
123                &SaveLessonRequest {
124                    session_id: req.session_id.as_deref(),
125                    project,
126                    topic_key: req.topic_key.as_deref(),
127                    title,
128                    content: &req.text,
129                    confidence: 0.7,
130                    source_evidence: None,
131                    files: files_json.as_deref(),
132                    branch: req.branch.as_deref(),
133                    scope,
134                    created_at_epoch: req.created_at_epoch,
135                    stale_after_epoch: None,
136                },
137                reference_time_epoch,
138            )?;
139            let mut logged_plan = operation_plan.clone();
140            logged_plan.target_memory_id = Some(id);
141            if logged_plan.op == MemoryLifecycleOp::Noop {
142                logged_plan.op = MemoryLifecycleOp::Update;
143                logged_plan.noop_reason = None;
144                logged_plan.reason =
145                    "existing lesson memory was reinforced by direct save".to_string();
146            }
147            crate::memory::operation::insert_operation_log(
148                conn,
149                &operation_input,
150                &logged_plan,
151                Some(id),
152            )?;
153            mark_direct_save_poisoning_metadata(conn, id, acknowledgement)?;
154            Ok((id, logged_plan.op))
155        })
156    } else {
157        crate::memory::operation::with_operation_savepoint(conn, || {
158            let (operation_input, operation_plan) = crate::memory::operation::plan_direct_save(
159                conn,
160                "direct",
161                "save_memory",
162                project,
163                scope,
164                memory_type,
165                effective_topic_key.as_deref(),
166                title,
167                &req.text,
168                files_json.as_deref(),
169                req.branch.as_deref(),
170                None,
171                None,
172            )?;
173            if operation_plan.op == MemoryLifecycleOp::Noop {
174                let id = operation_plan
175                    .target_memory_id
176                    .ok_or_else(|| anyhow!("noop memory operation missing existing memory id"))?;
177                crate::memory::operation::insert_operation_log(
178                    conn,
179                    &operation_input,
180                    &operation_plan,
181                    Some(id),
182                )?;
183                mark_direct_save_poisoning_metadata(conn, id, acknowledgement)?;
184                return Ok((id, MemoryLifecycleOp::Noop));
185            }
186            let previous_preference = if memory_type == "preference" {
187                operation_plan
188                    .target_memory_id
189                    .map(|memory_id| {
190                        conn.query_row(
191                            "SELECT content FROM memories WHERE id = ?1",
192                            [memory_id],
193                            |row| row.get::<_, String>(0),
194                        )
195                        .with_context(|| {
196                            format!("load preference before direct save update id={memory_id}")
197                        })
198                    })
199                    .transpose()?
200            } else {
201                None
202            };
203            let result = crate::memory::insert_memory_full_with_operation_log(
204                conn,
205                req.session_id.as_deref(),
206                project,
207                req.topic_key.as_deref(),
208                title,
209                &req.text,
210                memory_type,
211                files_json.as_deref(),
212                req.branch.as_deref(),
213                scope,
214                req.created_at_epoch,
215                reference_time_epoch,
216                &operation_input,
217                &operation_plan,
218            )?;
219            if let Some(previous_text) = previous_preference {
220                crate::memory::preference::compilation::enqueue_for_memory_ids(conn, &[result.0])?;
221                crate::memory::preference::reinforcement::reconcile_in_place_preference_update(
222                    conn,
223                    result.0,
224                    &previous_text,
225                    &req.text,
226                )?;
227            }
228            mark_direct_save_poisoning_metadata(conn, result.0, acknowledgement)?;
229            Ok(result)
230        })
231    };
232
233    let (id, operation) = match save_result {
234        Ok(result) => result,
235        Err(err) => {
236            if let Err(cleanup_err) = cleanup_local_copy(&local_copy) {
237                return Err(err.context(format!(
238                    "database save failed and local copy cleanup failed: {cleanup_err}"
239                )));
240            }
241            return Err(err);
242        }
243    };
244
245    discard_local_copy_backup(&local_copy);
246    let durable = load_durable_write_details(conn, id)?;
247    let local_copy_result = local_copy.result();
248    let claim_result = write_claim_after_durable_save(conn, id, req);
249
250    Ok(SaveMemoryResult {
251        id,
252        status: "saved".to_string(),
253        memory_type: durable.memory_type,
254        project: durable.project,
255        scope: durable.scope,
256        topic_key: durable.topic_key,
257        branch: durable.branch,
258        operation: operation.as_str().to_string(),
259        created_at_epoch: durable.created_at_epoch,
260        reference_time_epoch: durable.reference_time_epoch,
261        updated_at_epoch: durable.updated_at_epoch,
262        upserted: req.topic_key.is_some(),
263        local_status: local_copy_result.status.clone(),
264        local_path: local_copy_result.path.clone(),
265        local_copy: local_copy_result,
266        claim_status: claim_result.status,
267        claim_id: claim_result.id,
268        claim_error: claim_result.error,
269        next_step: SaveMemoryNextStep {
270            tool: "get_observations".to_string(),
271            ids: vec![id],
272            source: "memory".to_string(),
273            reason: format!(
274                "Verify the durable memory with get_observations(ids=[{id}], source='memory') or search(project='{}').",
275                durable_project_hint(project)
276            ),
277        },
278    })
279}
280
281fn direct_save_pattern_acknowledgement(
282    title: &str,
283    text: &str,
284    acknowledged_pattern_id: Option<&str>,
285) -> Result<Option<InstructionPatternMatch>> {
286    let scan_text = format!("{title}\n{text}");
287    let matched = scan_instruction_pattern(&scan_text);
288    let acknowledged_pattern_id = acknowledged_pattern_id
289        .map(str::trim)
290        .filter(|value| !value.is_empty());
291    match (matched, acknowledged_pattern_id) {
292        (Some(matched), Some(acknowledged)) if acknowledged == matched.pattern_id => {
293            Ok(Some(matched))
294        }
295        (Some(matched), Some(acknowledged)) => {
296            Err(SaveMemoryValidationError::new(format!(
297                "save_memory acknowledged pattern {acknowledged} does not match instruction-pattern {}@v{}",
298                matched.pattern_id, matched.pattern_set_version
299            ))
300            .into())
301        }
302        (Some(matched), None) => Err(SaveMemoryValidationError::new(format!(
303            "save_memory text matched instruction-pattern {}@v{}; review and acknowledge the pattern before saving",
304            matched.pattern_id, matched.pattern_set_version
305        ))
306        .into()),
307        (None, Some(acknowledged)) => Err(SaveMemoryValidationError::new(format!(
308            "save_memory acknowledge_pattern {acknowledged} was provided, but no instruction-pattern matched"
309        ))
310        .into()),
311        (None, None) => Ok(None),
312    }
313}
314
315fn mark_direct_save_poisoning_metadata(
316    conn: &Connection,
317    memory_id: i64,
318    acknowledgement: Option<InstructionPatternMatch>,
319) -> Result<()> {
320    if let Some(acknowledgement) = acknowledgement {
321        let now = chrono::Utc::now().timestamp();
322        conn.execute(
323            "UPDATE memories
324             SET source_trust_class = ?1,
325                 acknowledged_pattern_id = ?2,
326                 acknowledged_pattern_version = ?3,
327                 acknowledged_at_epoch = ?4
328             WHERE id = ?5",
329            rusqlite::params![
330                DIRECT_SAVE_TRUST_CLASS.as_str(),
331                acknowledgement.pattern_id,
332                acknowledgement.pattern_set_version,
333                now,
334                memory_id
335            ],
336        )?;
337    } else {
338        conn.execute(
339            "UPDATE memories SET source_trust_class = ?1 WHERE id = ?2",
340            rusqlite::params![DIRECT_SAVE_TRUST_CLASS.as_str(), memory_id],
341        )?;
342    }
343    Ok(())
344}
345
346struct ValidatedSaveMemoryRequest {
347    memory_type: String,
348    scope: String,
349}
350
351fn validate_save_memory_request(req: &SaveMemoryRequest) -> Result<ValidatedSaveMemoryRequest> {
352    if req.text.trim().is_empty() {
353        return Err(SaveMemoryValidationError::new("save_memory text must not be blank").into());
354    }
355
356    let memory_type = match req.memory_type.as_deref() {
357        Some(value) => {
358            let normalized = value.trim().to_ascii_lowercase();
359            let parsed = MemoryType::parse(&normalized).ok_or_else(|| {
360                SaveMemoryValidationError::new(format!(
361                    "save_memory memory_type must be one of: {}",
362                    MEMORY_TYPES.join(", ")
363                ))
364            })?;
365            parsed.as_str().to_string()
366        }
367        None => MemoryType::Discovery.as_str().to_string(),
368    };
369
370    let scope = match req.scope.as_deref() {
371        Some(value) => match value.trim().to_ascii_lowercase().as_str() {
372            "project" => "project".to_string(),
373            "global" => "global".to_string(),
374            _ => {
375                return Err(SaveMemoryValidationError::new(
376                    "save_memory scope must be one of: project, global",
377                )
378                .into());
379            }
380        },
381        None => "project".to_string(),
382    };
383
384    Ok(ValidatedSaveMemoryRequest { memory_type, scope })
385}
386
387struct ClaimSaveResult {
388    status: String,
389    id: Option<i64>,
390    error: Option<String>,
391}
392
393fn write_claim_after_durable_save(
394    conn: &Connection,
395    memory_id: i64,
396    req: &SaveMemoryRequest,
397) -> ClaimSaveResult {
398    if !claims_enabled(req.claim_enabled) {
399        return ClaimSaveResult {
400            status: "disabled".to_string(),
401            id: None,
402            error: None,
403        };
404    }
405
406    let claim_source = req.claim_source.as_deref().unwrap_or("manual_save");
407    match insert_memory_claim(
408        conn,
409        &ClaimWriteRequest {
410            memory_id,
411            session_id: req.session_id.as_deref(),
412            host: req.host.as_deref(),
413            claim_source,
414        },
415    ) {
416        Ok(claim_id) => ClaimSaveResult {
417            status: "saved".to_string(),
418            id: Some(claim_id),
419            error: None,
420        },
421        Err(err) => {
422            let error = format!("{err:#}");
423            crate::log::error(
424                "memory-claim",
425                &format!("claim write failed memory_id={} error={}", memory_id, error),
426            );
427            ClaimSaveResult {
428                status: "failed".to_string(),
429                id: None,
430                error: Some(error),
431            }
432        }
433    }
434}
435
436struct LocalCopyPlan {
437    status: String,
438    path: Option<PathBuf>,
439    reason: Option<String>,
440    content: Option<String>,
441    backup: Option<LocalCopyBackup>,
442}
443
444impl LocalCopyPlan {
445    fn result(&self) -> LocalCopyResult {
446        LocalCopyResult {
447            status: self.status.clone(),
448            path: self.path.as_ref().map(|path| path.display().to_string()),
449            reason: self.reason.clone(),
450        }
451    }
452}
453
454struct LocalCopyBackup {
455    restore_path: PathBuf,
456    backup_path: PathBuf,
457}
458
459fn prepare_local_copy(
460    project: &str,
461    title: &str,
462    req: &SaveMemoryRequest,
463) -> Result<LocalCopyPlan> {
464    if !local_copy_enabled_override(req.local_copy_enabled) {
465        return Ok(LocalCopyPlan {
466            status: "disabled".to_string(),
467            path: None,
468            reason: Some("local copy disabled by request or configuration".to_string()),
469            content: None,
470            backup: None,
471        });
472    }
473
474    let local_path =
475        resolve_local_note_path(project, req.title.as_deref(), req.local_path.as_deref())?;
476    let content = build_local_note_content(project, title, &req.text);
477    Ok(LocalCopyPlan {
478        status: "saved".to_string(),
479        path: Some(local_path),
480        reason: None,
481        content: Some(content),
482        backup: None,
483    })
484}
485
486fn effective_topic_key(req: &SaveMemoryRequest, memory_type: &str) -> Option<String> {
487    if memory_type == "lesson" {
488        return req.topic_key.clone().or_else(|| {
489            Some(format!(
490                "lesson-{}",
491                crate::memory::slugify_for_topic(&req.text, 64)
492            ))
493        });
494    }
495    req.topic_key.clone()
496}
497
498struct DurableWriteDetails {
499    project: String,
500    scope: String,
501    topic_key: Option<String>,
502    branch: Option<String>,
503    memory_type: String,
504    created_at_epoch: i64,
505    reference_time_epoch: i64,
506    updated_at_epoch: i64,
507}
508
509fn load_durable_write_details(conn: &Connection, id: i64) -> Result<DurableWriteDetails> {
510    conn.query_row(
511        "SELECT project, COALESCE(scope, 'project'), topic_key, branch, memory_type,
512                created_at_epoch, COALESCE(reference_time_epoch, created_at_epoch), updated_at_epoch
513         FROM memories
514         WHERE id = ?1",
515        [id],
516        |row| {
517            Ok(DurableWriteDetails {
518                project: row.get(0)?,
519                scope: row.get(1)?,
520                topic_key: row.get(2)?,
521                branch: row.get(3)?,
522                memory_type: row.get(4)?,
523                created_at_epoch: row.get(5)?,
524                reference_time_epoch: row.get(6)?,
525                updated_at_epoch: row.get(7)?,
526            })
527        },
528    )
529    .with_context(|| format!("load durable write details for memory {id}"))
530}
531
532fn durable_project_hint(project: &str) -> String {
533    project.replace('\'', "\\'")
534}
535
536fn write_local_copy(local_copy: &mut LocalCopyPlan) -> Result<()> {
537    if let (Some(path), Some(content)) = (local_copy.path.as_deref(), local_copy.content.as_deref())
538    {
539        let backup = backup_existing_local_copy(path)?;
540        if let Err(err) = write_local_note(path, content) {
541            if let Err(restore_err) = restore_local_copy(backup.as_ref()) {
542                return Err(err.context(format!(
543                    "write local copy failed and restore failed: {restore_err}"
544                )));
545            }
546            return Err(err);
547        }
548        local_copy.backup = backup;
549    }
550    Ok(())
551}
552
553fn cleanup_local_copy(local_copy: &LocalCopyPlan) -> Result<()> {
554    restore_local_copy(local_copy.backup.as_ref())?;
555    match (local_copy.path.as_deref(), local_copy.backup.as_ref()) {
556        (Some(path), None) => remove_local_copy_file(path),
557        _ => Ok(()),
558    }
559}
560
561fn backup_existing_local_copy(path: &Path) -> Result<Option<LocalCopyBackup>> {
562    match std::fs::symlink_metadata(path) {
563        Ok(metadata) => {
564            let restore_path = backup_restore_path(path, &metadata)?;
565            let backup_path = allocate_backup_path(&restore_path);
566            std::fs::rename(&restore_path, &backup_path).with_context(|| {
567                format!(
568                    "move existing local copy {} to backup {}",
569                    restore_path.display(),
570                    backup_path.display()
571                )
572            })?;
573            Ok(Some(LocalCopyBackup {
574                restore_path,
575                backup_path,
576            }))
577        }
578        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
579        Err(err) => Err(anyhow!(
580            "check existing local copy at {}: {err}",
581            path.display()
582        )),
583    }
584}
585
586fn backup_restore_path(path: &Path, metadata: &std::fs::Metadata) -> Result<PathBuf> {
587    let file_type = metadata.file_type();
588    if file_type.is_dir() {
589        return Err(anyhow!(
590            "local_path {} must reference a file, not a directory",
591            path.display()
592        ));
593    }
594
595    if file_type.is_symlink() {
596        let target_path = path
597            .canonicalize()
598            .with_context(|| format!("resolve local_path symlink target at {}", path.display()))?;
599        if target_path.is_dir() {
600            return Err(anyhow!(
601                "local_path {} must reference a file, not a directory",
602                path.display()
603            ));
604        }
605        return Ok(target_path);
606    }
607
608    Ok(path.to_path_buf())
609}
610
611fn restore_local_copy(backup: Option<&LocalCopyBackup>) -> Result<()> {
612    if let Some(backup) = backup {
613        remove_local_copy_file(&backup.restore_path)?;
614        std::fs::rename(&backup.backup_path, &backup.restore_path).with_context(|| {
615            format!(
616                "restore local copy from backup {} to {}",
617                backup.backup_path.display(),
618                backup.restore_path.display()
619            )
620        })?;
621    }
622    Ok(())
623}
624
625fn remove_local_copy_file(path: &Path) -> Result<()> {
626    match std::fs::remove_file(path) {
627        Ok(()) => Ok(()),
628        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
629        Err(err) => Err(err).with_context(|| format!("remove local copy at {}", path.display())),
630    }
631}
632
633fn discard_local_copy_backup(local_copy: &LocalCopyPlan) {
634    if let Some(backup) = local_copy.backup.as_ref() {
635        let _ = std::fs::remove_file(&backup.backup_path);
636    }
637}
638
639fn allocate_backup_path(path: &Path) -> PathBuf {
640    let parent = path.parent().unwrap_or_else(|| Path::new("."));
641    let file_name = path
642        .file_name()
643        .and_then(|name| name.to_str())
644        .unwrap_or("local-copy");
645    let timestamp = SystemTime::now()
646        .duration_since(UNIX_EPOCH)
647        .unwrap_or_default()
648        .as_nanos();
649    parent.join(format!(
650        ".{file_name}.remem-backup-{}-{timestamp}.tmp",
651        std::process::id()
652    ))
653}