graphrefly/adapters/
agentic_memory_storage.rs

1//! Graph-bound persistence sidecar for agentic-memory records (D172).
2//!
3//! This adapter composes D166 strict record frames with passive storage tiers.
4//! It never restores a live graph and never adds storage fields to records.
5//! The v1 sidecar assumes a single writer per chosen storage prefix/change log;
6//! corrupt, sparse, or non-contiguous durable input fails honestly during load.
7
8use std::cell::{Cell, RefCell};
9use std::collections::HashSet;
10use std::rc::Rc;
11
12use serde_json::{json, Value};
13
14use crate::graph::{Graph, GraphNodeOpts};
15use crate::json::{strict_canonical_json_bytes, strict_json_decode, Codec, JsonValue};
16use crate::node::Node;
17use crate::protocol::Message;
18use crate::solutions::{
19    agentic_memory_record_frame, agentic_memory_record_frame_codec, AgenticMemoryRecord,
20    AgenticMemoryRecordFrame,
21};
22use crate::storage::{
23    AppendLogReadOptions, AppendLogStorageTier, KvStorageTier, StorageError, StorageResult,
24};
25
26type Disposer = Box<dyn FnOnce()>;
27
28/// `constant` constant.
29pub const AGENTIC_MEMORY_RECORD_SNAPSHOT_FORMAT: &str = "graphrefly.agenticMemory.records.snapshot";
30/// `constant` constant.
31pub const AGENTIC_MEMORY_RECORD_CHANGE_FORMAT: &str = "graphrefly.agenticMemory.records.change";
32/// `constant` constant.
33pub const AGENTIC_MEMORY_RECORD_STORAGE_FRAME_VERSION: u32 = 1;
34
35#[derive(Clone, Debug, Eq, PartialEq)]
36/// `AgenticMemoryRecordsPersistenceStatus` variants.
37pub enum AgenticMemoryRecordsPersistenceStatus {
38    /// `Starting` variant.
39    Starting,
40    /// `Ready` variant.
41    Ready,
42    /// `Flushing` variant.
43    Flushing,
44    /// `Errored` variant.
45    Errored,
46    /// `Disposed` variant.
47    Disposed,
48}
49
50#[derive(Clone, Debug, Eq, PartialEq)]
51/// `AgenticMemoryRecordsPersistenceCursor` data container.
52pub struct AgenticMemoryRecordsPersistenceCursor {
53    /// `change_seq` field for change seq.
54    pub change_seq: Option<u64>,
55    /// `snapshot_writes` field for snapshot writes.
56    pub snapshot_writes: usize,
57    /// `change_writes` field for change writes.
58    pub change_writes: usize,
59}
60
61#[derive(Clone, Debug, Eq, PartialEq)]
62/// `AgenticMemoryRecordsPersistenceStatusFact` data container.
63pub struct AgenticMemoryRecordsPersistenceStatusFact {
64    /// `state` field for state.
65    pub state: AgenticMemoryRecordsPersistenceStatus,
66    /// `pending` field for pending.
67    pub pending: usize,
68    /// `writes` field for writes.
69    pub writes: usize,
70    /// `errors` field for errors.
71    pub errors: usize,
72    /// `cursor` field for cursor.
73    pub cursor: AgenticMemoryRecordsPersistenceCursor,
74}
75
76#[derive(Clone, Debug, Eq, PartialEq)]
77/// `AgenticMemoryRecordsPersistenceErrorFact` data container.
78pub struct AgenticMemoryRecordsPersistenceErrorFact {
79    /// `phase` field for phase.
80    pub phase: String,
81    /// `message` field for message.
82    pub message: String,
83    /// `cursor` field for cursor.
84    pub cursor: AgenticMemoryRecordsPersistenceCursor,
85}
86
87#[derive(Clone, Debug, PartialEq)]
88/// `AgenticMemoryRecordsRestoreState` data container.
89pub struct AgenticMemoryRecordsRestoreState {
90    /// `records` field for records.
91    pub records: Vec<AgenticMemoryRecord<JsonValue>>,
92    /// `snapshot_found` field for snapshot found.
93    pub snapshot_found: bool,
94    /// `changes_applied` field for changes applied.
95    pub changes_applied: usize,
96    /// `cursor` field for cursor.
97    pub cursor: Option<u64>,
98    /// `source` field for source.
99    pub source: String,
100}
101
102#[derive(Clone, Default)]
103/// `LoadAgenticMemoryRecordsStateOptions` data container.
104pub struct LoadAgenticMemoryRecordsStateOptions<'a> {
105    /// `storage_prefix` field for storage prefix.
106    pub storage_prefix: Option<&'a str>,
107    /// `snapshot_key` field for snapshot key.
108    pub snapshot_key: Option<&'a str>,
109    /// `change_log` field for change log.
110    pub change_log: Option<&'a dyn AppendLogStorageTier<Value>>,
111}
112
113#[derive(Clone)]
114/// `PersistAgenticMemoryRecordsOptions` data container.
115pub struct PersistAgenticMemoryRecordsOptions {
116    /// `graph` field for graph.
117    pub graph: Option<Graph>,
118    /// `name` field for name.
119    pub name: Option<String>,
120    /// `storage_prefix` field for storage prefix.
121    pub storage_prefix: Option<String>,
122    /// `snapshot_key` field for snapshot key.
123    pub snapshot_key: Option<String>,
124    /// `snapshot_store` field for snapshot store.
125    pub snapshot_store: Rc<dyn KvStorageTier<Value>>,
126    /// Optional append-only change log.
127    ///
128    /// D172 v1 assumes one writer per selected log/prefix. Multi-writer merge,
129    /// fencing, and conflict repair belong in a future adapter decision; this
130    /// sidecar loads only contiguous frames and reports gaps as corruption.
131    pub change_log: Option<Rc<dyn AppendLogStorageTier<Value>>>,
132    /// `snapshot_on_attach` field for snapshot on attach.
133    pub snapshot_on_attach: bool,
134}
135
136impl PersistAgenticMemoryRecordsOptions {
137    /// Creates or computes `new`.
138    pub fn new(
139        snapshot_store: Rc<dyn KvStorageTier<Value>>,
140        storage_prefix: impl Into<String>,
141    ) -> Self {
142        Self {
143            graph: None,
144            name: None,
145            storage_prefix: Some(storage_prefix.into()),
146            snapshot_key: None,
147            snapshot_store,
148            change_log: None,
149            snapshot_on_attach: true,
150        }
151    }
152}
153
154/// `AgenticMemoryRecordsPersistence` data container.
155pub struct AgenticMemoryRecordsPersistence {
156    /// `ready` field for ready.
157    pub ready: Node<bool>,
158    /// `status` field for status.
159    pub status: Node<Value>,
160    /// `error` field for error.
161    pub error: Node<Value>,
162    /// `cursor` field for cursor.
163    pub cursor: Node<Value>,
164    flush: Rc<dyn Fn() -> StorageResult<()>>,
165    snapshot: Rc<dyn Fn() -> StorageResult<()>>,
166    dispose: RefCell<Option<Disposer>>,
167}
168
169impl AgenticMemoryRecordsPersistence {
170    /// Updates or reads `flush`.
171    pub fn flush(&self) -> StorageResult<()> {
172        (self.flush)()
173    }
174
175    /// Updates or reads `snapshot`.
176    pub fn snapshot(&self) -> StorageResult<()> {
177        (self.snapshot)()
178    }
179
180    /// Updates or reads `dispose`.
181    pub fn dispose(&self) {
182        if let Some(dispose) = self.dispose.borrow_mut().take() {
183            dispose();
184        }
185    }
186
187    /// Updates or reads `cursor_fact`.
188    pub fn cursor_fact(&self) -> StorageResult<AgenticMemoryRecordsPersistenceCursor> {
189        parse_cursor_fact(
190            &self.cursor.cache().ok_or_else(|| {
191                StorageError::backend("agenticMemoryRecords cursor fact is absent")
192            })?,
193        )
194    }
195}
196
197impl Drop for AgenticMemoryRecordsPersistence {
198    fn drop(&mut self) {
199        if let Some(dispose) = self.dispose.borrow_mut().take() {
200            dispose();
201        }
202    }
203}
204
205/// `OpenPersistentAgenticMemoryRecords` data container.
206pub struct OpenPersistentAgenticMemoryRecords {
207    /// `records` field for records.
208    pub records: Node<Vec<AgenticMemoryRecord<JsonValue>>>,
209    /// `persistence` field for persistence.
210    pub persistence: AgenticMemoryRecordsPersistence,
211    /// `loaded` field for loaded.
212    pub loaded: AgenticMemoryRecordsRestoreState,
213}
214
215/// `OpenPersistentAgenticMemoryRecordsOptions` data container.
216pub struct OpenPersistentAgenticMemoryRecordsOptions {
217    /// `graph` field for graph.
218    pub graph: Graph,
219    /// `name` field for name.
220    pub name: Option<String>,
221    /// `initial` field for initial.
222    pub initial: Vec<AgenticMemoryRecord<JsonValue>>,
223    /// `persistence` field for persistence.
224    pub persistence: PersistAgenticMemoryRecordsOptions,
225}
226
227/// Creates or computes `agentic_memory_records_snapshot_key`.
228pub fn agentic_memory_records_snapshot_key(storage_prefix: &str) -> StorageResult<String> {
229    if storage_prefix.is_empty() {
230        return Err(StorageError::backend("storage_prefix must be non-empty"));
231    }
232    Ok(format!("{storage_prefix}/records.snapshot"))
233}
234
235/// Creates or computes `agentic_memory_record_snapshot_frame`.
236pub fn agentic_memory_record_snapshot_frame(
237    records: &[AgenticMemoryRecord<JsonValue>],
238    change_cursor: Option<u64>,
239) -> StorageResult<Value> {
240    let cursor = change_cursor
241        .map(|seq| i64::try_from(seq).map_err(|_| StorageError::backend("change cursor overflow")))
242        .transpose()?
243        .unwrap_or(-1);
244    let records = records
245        .iter()
246        .map(record_frame_json)
247        .collect::<StorageResult<Vec<_>>>()?;
248    Ok(json!({
249        "format": AGENTIC_MEMORY_RECORD_SNAPSHOT_FORMAT,
250        "version": AGENTIC_MEMORY_RECORD_STORAGE_FRAME_VERSION,
251        "changeCursor": cursor,
252        "records": records,
253    }))
254}
255
256/// Creates or computes `agentic_memory_record_change_frame`.
257pub fn agentic_memory_record_change_frame(
258    records: &[AgenticMemoryRecord<JsonValue>],
259) -> StorageResult<Value> {
260    let records = records
261        .iter()
262        .map(record_frame_json)
263        .collect::<StorageResult<Vec<_>>>()?;
264    Ok(json!({
265        "format": AGENTIC_MEMORY_RECORD_CHANGE_FORMAT,
266        "version": AGENTIC_MEMORY_RECORD_STORAGE_FRAME_VERSION,
267        "change": { "kind": "replaceAll", "records": records },
268    }))
269}
270
271/// Creates or computes `load_agentic_memory_records_state`.
272pub fn load_agentic_memory_records_state(
273    snapshot_store: &dyn KvStorageTier<Value>,
274    options: LoadAgenticMemoryRecordsStateOptions<'_>,
275) -> StorageResult<AgenticMemoryRecordsRestoreState> {
276    let storage_prefix = options.storage_prefix.unwrap_or("agentic-memory");
277    let snapshot_key = options
278        .snapshot_key
279        .map(ToOwned::to_owned)
280        .unwrap_or(agentic_memory_records_snapshot_key(storage_prefix)?);
281    let snapshot = snapshot_store.get(&snapshot_key)?;
282    let mut records = Vec::new();
283    let mut snapshot_found = false;
284    let mut cursor = None;
285    if let Some(snapshot) = snapshot {
286        let parsed = parse_snapshot_frame(&snapshot)?;
287        records = parsed.records;
288        snapshot_found = true;
289        cursor = parsed.change_cursor;
290    }
291    let mut changes_applied = 0usize;
292    if let Some(change_log) = options.change_log {
293        let entries = change_log.read(AppendLogReadOptions::default())?;
294        let mut expected = cursor.map(|seq| seq.saturating_add(1)).unwrap_or(0);
295        for entry in entries {
296            if let Some(snapshot_cursor) = cursor {
297                if entry.seq <= snapshot_cursor {
298                    continue;
299                }
300            }
301            if entry.seq != expected {
302                return Err(StorageError::backend(
303                    "agentic memory records change log is non-contiguous",
304                ));
305            }
306            let change = parse_change_frame(&entry.value)?;
307            records = change;
308            cursor = Some(entry.seq);
309            expected = entry.seq.saturating_add(1);
310            changes_applied += 1;
311        }
312    }
313    Ok(AgenticMemoryRecordsRestoreState {
314        records,
315        snapshot_found,
316        changes_applied,
317        cursor,
318        source: match (snapshot_found, changes_applied > 0) {
319            (true, true) => "snapshot+changes",
320            (true, false) => "snapshot",
321            (false, true) => "changes",
322            (false, false) => "empty",
323        }
324        .to_owned(),
325    })
326}
327
328/// Creates or computes `persist_agentic_memory_records`.
329pub fn persist_agentic_memory_records(
330    records: &Node<Vec<AgenticMemoryRecord<JsonValue>>>,
331    options: PersistAgenticMemoryRecordsOptions,
332) -> StorageResult<AgenticMemoryRecordsPersistence> {
333    persist_agentic_memory_records_with_cursor(records, options, None)
334}
335
336fn persist_agentic_memory_records_with_cursor(
337    records: &Node<Vec<AgenticMemoryRecord<JsonValue>>>,
338    options: PersistAgenticMemoryRecordsOptions,
339    initial_change_seq: Option<u64>,
340) -> StorageResult<AgenticMemoryRecordsPersistence> {
341    let graph = options.graph.as_ref().ok_or_else(|| {
342        StorageError::backend(
343            "persistAgenticMemoryRecords: graph is required for graph-visible sidecar facts",
344        )
345    })?;
346    if !graph.contains_core(&records.erased()) {
347        return Err(StorageError::backend(
348            "persistAgenticMemoryRecords: records node belongs to a different graph or is not graph-registered",
349        ));
350    }
351    let prefix = options
352        .name
353        .clone()
354        .or_else(|| options.storage_prefix.clone())
355        .unwrap_or_else(|| "agenticMemoryRecords.persistence".to_owned());
356    let snapshot_key = options
357        .snapshot_key
358        .clone()
359        .unwrap_or(agentic_memory_records_snapshot_key(
360            options.storage_prefix.as_deref().unwrap_or(&prefix),
361        )?);
362    let ready = graph.state_opts(false, GraphNodeOpts::named(format!("{prefix}.ready")));
363    let status = graph.state_opts(
364        status_json(
365            AgenticMemoryRecordsPersistenceStatus::Starting,
366            0,
367            0,
368            cursor_json(initial_change_seq, 0, 0),
369        ),
370        GraphNodeOpts::named(format!("{prefix}.status")),
371    );
372    let error = graph.state_opts(Value::Null, GraphNodeOpts::named(format!("{prefix}.error")));
373    let cursor = graph.state_opts(
374        cursor_json(initial_change_seq, 0, 0),
375        GraphNodeOpts::named(format!("{prefix}.cursor")),
376    );
377    let latest = Rc::new(RefCell::new(records.cache().unwrap_or_default()));
378    let change_seq = Rc::new(Cell::new(initial_change_seq));
379    let snapshot_writes = Rc::new(Cell::new(0usize));
380    let change_writes = Rc::new(Cell::new(0usize));
381    let error_count = Rc::new(Cell::new(0usize));
382    let disposed = Rc::new(Cell::new(false));
383    let failed_write = Rc::new(RefCell::new(None::<StorageError>));
384    let snapshot_store = options.snapshot_store.clone();
385    let change_log = options.change_log.clone();
386
387    let write_snapshot = {
388        let latest = latest.clone();
389        let snapshot_store = snapshot_store.clone();
390        let snapshot_key = snapshot_key.clone();
391        let change_seq = change_seq.clone();
392        let snapshot_writes = snapshot_writes.clone();
393        let change_writes = change_writes.clone();
394        let ready = ready.clone();
395        let status = status.clone();
396        let error = error.clone();
397        let cursor = cursor.clone();
398        let failed_write = failed_write.clone();
399        move || {
400            let frame = agentic_memory_record_snapshot_frame(&latest.borrow(), change_seq.get())?;
401            snapshot_store.set(&snapshot_key, frame)?;
402            failed_write.replace(None);
403            snapshot_writes.set(snapshot_writes.get().saturating_add(1));
404            let cursor_value =
405                cursor_json(change_seq.get(), snapshot_writes.get(), change_writes.get());
406            ready.set(true);
407            status.set(status_json(
408                AgenticMemoryRecordsPersistenceStatus::Ready,
409                0,
410                0,
411                cursor_value.clone(),
412            ));
413            error.set(Value::Null);
414            cursor.set(cursor_value);
415            Ok(())
416        }
417    };
418
419    if options.snapshot_on_attach {
420        write_snapshot()?;
421    } else {
422        ready.set(true);
423        status.set(status_json(
424            AgenticMemoryRecordsPersistenceStatus::Ready,
425            0,
426            0,
427            cursor_json(initial_change_seq, 0, 0),
428        ));
429    }
430
431    let write_snapshot_for_control = Rc::new(write_snapshot);
432    let subscribing = Rc::new(Cell::new(true));
433    let unsubscribe = {
434        let latest = latest.clone();
435        let change_log = change_log.clone();
436        let change_seq = change_seq.clone();
437        let snapshot_writes = snapshot_writes.clone();
438        let change_writes = change_writes.clone();
439        let ready = ready.clone();
440        let status = status.clone();
441        let error = error.clone();
442        let cursor = cursor.clone();
443        let error_count = error_count.clone();
444        let failed_write = failed_write.clone();
445        let subscribing = subscribing.clone();
446        records.subscribe(move |message| {
447            let Message::Data(next) = message else {
448                return;
449            };
450            let Some(next) = next
451                .as_ref()
452                .downcast_ref::<Vec<AgenticMemoryRecord<JsonValue>>>()
453            else {
454                return;
455            };
456            latest.replace(next.clone());
457            if subscribing.get() {
458                subscribing.set(false);
459                return;
460            }
461            let Some(change_log) = &change_log else {
462                return;
463            };
464            match agentic_memory_record_change_frame(next)
465                .and_then(|frame| change_log.append(frame).map(|entry| entry.seq))
466            {
467                Ok(seq) => {
468                    failed_write.replace(None);
469                    change_seq.set(Some(seq));
470                    change_writes.set(change_writes.get().saturating_add(1));
471                    let cursor_value =
472                        cursor_json(change_seq.get(), snapshot_writes.get(), change_writes.get());
473                    ready.set(true);
474                    status.set(status_json(
475                        AgenticMemoryRecordsPersistenceStatus::Ready,
476                        0,
477                        error_count.get(),
478                        cursor_value.clone(),
479                    ));
480                    error.set(Value::Null);
481                    cursor.set(cursor_value);
482                }
483                Err(err) => {
484                    failed_write.replace(Some(err.clone()));
485                    error_count.set(error_count.get().saturating_add(1));
486                    ready.set(false);
487                    let cursor_value =
488                        cursor_json(change_seq.get(), snapshot_writes.get(), change_writes.get());
489                    status.set(status_json(
490                        AgenticMemoryRecordsPersistenceStatus::Errored,
491                        0,
492                        error_count.get(),
493                        cursor_value.clone(),
494                    ));
495                    error.set(error_json("change", &err, cursor_value));
496                }
497            }
498        })
499    };
500    subscribing.set(false);
501    let flush = {
502        let status = status.clone();
503        let cursor = cursor.clone();
504        let ready = ready.clone();
505        let change_seq = change_seq.clone();
506        let snapshot_writes = snapshot_writes.clone();
507        let change_writes = change_writes.clone();
508        let error_count = error_count.clone();
509        let disposed = disposed.clone();
510        let error = error.clone();
511        let failed_write = failed_write.clone();
512        Rc::new(move || {
513            if disposed.get() {
514                return Ok(());
515            }
516            let cursor_value =
517                cursor_json(change_seq.get(), snapshot_writes.get(), change_writes.get());
518            if let Some(err) = failed_write.borrow().clone() {
519                ready.set(false);
520                status.set(status_json(
521                    AgenticMemoryRecordsPersistenceStatus::Errored,
522                    0,
523                    error_count.get(),
524                    cursor_value.clone(),
525                ));
526                error.set(error_json("flush", &err, cursor_value));
527                return Err(err);
528            }
529            ready.set(true);
530            status.set(status_json(
531                AgenticMemoryRecordsPersistenceStatus::Ready,
532                0,
533                error_count.get(),
534                cursor_value.clone(),
535            ));
536            cursor.set(cursor_value);
537            Ok(())
538        })
539    };
540    let snapshot = {
541        let write_snapshot = write_snapshot_for_control.clone();
542        let disposed = disposed.clone();
543        Rc::new(move || {
544            if disposed.get() {
545                return Err(StorageError::backend(
546                    "agenticMemoryRecords persistence is disposed",
547                ));
548            }
549            write_snapshot()
550        })
551    };
552    let dispose = {
553        let disposed = disposed.clone();
554        let ready = ready.clone();
555        let status = status.clone();
556        let cursor = cursor.clone();
557        let change_seq = change_seq.clone();
558        let snapshot_writes = snapshot_writes.clone();
559        let change_writes = change_writes.clone();
560        Box::new(move || {
561            disposed.set(true);
562            unsubscribe();
563            ready.set(false);
564            status.set(status_json(
565                AgenticMemoryRecordsPersistenceStatus::Disposed,
566                0,
567                0,
568                cursor_json(change_seq.get(), snapshot_writes.get(), change_writes.get()),
569            ));
570            cursor.set(cursor_json(
571                change_seq.get(),
572                snapshot_writes.get(),
573                change_writes.get(),
574            ));
575        }) as Disposer
576    };
577
578    Ok(AgenticMemoryRecordsPersistence {
579        ready,
580        status,
581        error,
582        cursor,
583        flush,
584        snapshot,
585        dispose: RefCell::new(Some(dispose)),
586    })
587}
588
589/// Creates or computes `open_persistent_agentic_memory_records`.
590pub fn open_persistent_agentic_memory_records(
591    options: OpenPersistentAgenticMemoryRecordsOptions,
592) -> StorageResult<OpenPersistentAgenticMemoryRecords> {
593    let snapshot_key =
594        options
595            .persistence
596            .snapshot_key
597            .clone()
598            .or(agentic_memory_records_snapshot_key(
599                options
600                    .persistence
601                    .storage_prefix
602                    .as_deref()
603                    .unwrap_or("agentic-memory"),
604            )
605            .ok());
606    let mut loaded = load_agentic_memory_records_state(
607        options.persistence.snapshot_store.as_ref(),
608        LoadAgenticMemoryRecordsStateOptions {
609            storage_prefix: options.persistence.storage_prefix.as_deref(),
610            snapshot_key: snapshot_key.as_deref(),
611            change_log: options.persistence.change_log.as_deref(),
612        },
613    )?;
614    if loaded.source == "empty" {
615        loaded.records = options.initial;
616    }
617    let records = options.graph.state_opts(
618        loaded.records.clone(),
619        GraphNodeOpts::named(
620            options
621                .name
622                .clone()
623                .unwrap_or_else(|| "agenticMemoryRecords".to_owned()),
624        ),
625    );
626    let mut persistence_options = options.persistence;
627    if persistence_options.graph.is_none() {
628        persistence_options.graph = Some(options.graph.clone());
629    }
630    let persistence =
631        persist_agentic_memory_records_with_cursor(&records, persistence_options, loaded.cursor)?;
632    Ok(OpenPersistentAgenticMemoryRecords {
633        records,
634        persistence,
635        loaded,
636    })
637}
638
639struct ParsedSnapshot {
640    records: Vec<AgenticMemoryRecord<JsonValue>>,
641    change_cursor: Option<u64>,
642}
643
644fn parse_snapshot_frame(value: &Value) -> StorageResult<ParsedSnapshot> {
645    let object = value
646        .as_object()
647        .ok_or_else(|| StorageError::backend("agentic memory snapshot frame must be an object"))?;
648    assert_keys(
649        object.keys().map(String::as_str),
650        ["changeCursor", "format", "records", "version"],
651    )?;
652    if object.get("format")
653        != Some(&Value::String(
654            AGENTIC_MEMORY_RECORD_SNAPSHOT_FORMAT.to_owned(),
655        ))
656    {
657        return Err(StorageError::backend(
658            "agentic memory snapshot frame: invalid format",
659        ));
660    }
661    if object.get("version").and_then(Value::as_u64)
662        != Some(AGENTIC_MEMORY_RECORD_STORAGE_FRAME_VERSION as u64)
663    {
664        return Err(StorageError::backend(
665            "agentic memory snapshot frame: invalid version",
666        ));
667    }
668    let change_cursor = cursor_from_json(object.get("changeCursor"))?;
669    let records = object
670        .get("records")
671        .and_then(Value::as_array)
672        .ok_or_else(|| {
673            StorageError::backend("agentic memory snapshot frame: records must be an array")
674        })?
675        .iter()
676        .map(record_from_frame_json)
677        .collect::<StorageResult<Vec<_>>>()?;
678    validate_unique_record_set(&records, "agentic memory snapshot frame")?;
679    Ok(ParsedSnapshot {
680        records,
681        change_cursor,
682    })
683}
684
685fn parse_change_frame(value: &Value) -> StorageResult<Vec<AgenticMemoryRecord<JsonValue>>> {
686    let object = value
687        .as_object()
688        .ok_or_else(|| StorageError::backend("agentic memory change frame must be an object"))?;
689    assert_keys(
690        object.keys().map(String::as_str),
691        ["change", "format", "version"],
692    )?;
693    if object.get("format")
694        != Some(&Value::String(
695            AGENTIC_MEMORY_RECORD_CHANGE_FORMAT.to_owned(),
696        ))
697    {
698        return Err(StorageError::backend(
699            "agentic memory change frame: invalid format",
700        ));
701    }
702    if object.get("version").and_then(Value::as_u64)
703        != Some(AGENTIC_MEMORY_RECORD_STORAGE_FRAME_VERSION as u64)
704    {
705        return Err(StorageError::backend(
706            "agentic memory change frame: invalid version",
707        ));
708    }
709    let change = object
710        .get("change")
711        .and_then(Value::as_object)
712        .ok_or_else(|| {
713            StorageError::backend("agentic memory change frame: change must be an object")
714        })?;
715    assert_keys(change.keys().map(String::as_str), ["kind", "records"])?;
716    if change.get("kind") != Some(&Value::String("replaceAll".to_owned())) {
717        return Err(StorageError::backend(
718            "agentic memory change frame: change.kind must be replaceAll",
719        ));
720    }
721    let records = change
722        .get("records")
723        .and_then(Value::as_array)
724        .ok_or_else(|| {
725            StorageError::backend("agentic memory change frame: records must be an array")
726        })?
727        .iter()
728        .map(record_from_frame_json)
729        .collect::<StorageResult<Vec<_>>>()?;
730    validate_unique_record_set(&records, "agentic memory change frame")?;
731    Ok(records)
732}
733
734fn validate_unique_record_set(
735    records: &[AgenticMemoryRecord<JsonValue>],
736    context: &str,
737) -> StorageResult<()> {
738    let mut record_ids = HashSet::new();
739    let mut fragment_ids = HashSet::new();
740    for record in records {
741        if !record_ids.insert(record.id.as_str()) {
742            return Err(StorageError::backend(format!(
743                "{context}: duplicate record id '{}'",
744                record.id
745            )));
746        }
747        if !fragment_ids.insert(record.fragment.id.as_str()) {
748            return Err(StorageError::backend(format!(
749                "{context}: duplicate fragment id '{}'",
750                record.fragment.id
751            )));
752        }
753    }
754    Ok(())
755}
756
757fn record_frame_json(record: &AgenticMemoryRecord<JsonValue>) -> StorageResult<Value> {
758    let codec = agentic_memory_record_frame_codec();
759    let bytes = codec
760        .encode(&agentic_memory_record_frame(record.clone()))
761        .map_err(storage_json_error)?;
762    strict_json_decode(&bytes).map_err(storage_json_error)
763}
764
765fn record_from_frame_json(value: &Value) -> StorageResult<AgenticMemoryRecord<JsonValue>> {
766    strict_canonical_json_bytes(value)
767        .and_then(|bytes| agentic_memory_record_frame_codec().decode(&bytes))
768        .map(|frame: AgenticMemoryRecordFrame| frame.record)
769        .map_err(storage_json_error)
770}
771
772fn cursor_from_json(value: Option<&Value>) -> StorageResult<Option<u64>> {
773    let raw = value
774        .and_then(Value::as_i64)
775        .ok_or_else(|| StorageError::backend("changeCursor must be an integer"))?;
776    if raw < -1 {
777        return Err(StorageError::backend("changeCursor must be >= -1"));
778    }
779    Ok(if raw == -1 { None } else { Some(raw as u64) })
780}
781
782fn cursor_json(change_seq: Option<u64>, snapshot_writes: usize, change_writes: usize) -> Value {
783    json!({
784        "kind": "persistence.cursor",
785        "changeSeq": change_seq,
786        "snapshotWrites": snapshot_writes,
787        "changeWrites": change_writes,
788    })
789}
790
791fn status_json(
792    state: AgenticMemoryRecordsPersistenceStatus,
793    pending: usize,
794    errors: usize,
795    cursor: Value,
796) -> Value {
797    json!({
798        "state": match state {
799            AgenticMemoryRecordsPersistenceStatus::Starting => "starting",
800            AgenticMemoryRecordsPersistenceStatus::Ready => "ready",
801            AgenticMemoryRecordsPersistenceStatus::Flushing => "flushing",
802            AgenticMemoryRecordsPersistenceStatus::Errored => "errored",
803            AgenticMemoryRecordsPersistenceStatus::Disposed => "disposed",
804        },
805        "pending": pending,
806        "writes": cursor.get("snapshotWrites").and_then(Value::as_u64).unwrap_or(0)
807            + cursor.get("changeWrites").and_then(Value::as_u64).unwrap_or(0),
808        "errors": errors,
809        "cursor": cursor,
810    })
811}
812
813fn error_json(phase: &str, err: &StorageError, cursor: Value) -> Value {
814    json!({
815        "phase": phase,
816        "message": err.to_string(),
817        "cursor": cursor,
818    })
819}
820
821fn parse_cursor_fact(value: &Value) -> StorageResult<AgenticMemoryRecordsPersistenceCursor> {
822    Ok(AgenticMemoryRecordsPersistenceCursor {
823        change_seq: value.get("changeSeq").and_then(Value::as_u64),
824        snapshot_writes: value
825            .get("snapshotWrites")
826            .and_then(Value::as_u64)
827            .unwrap_or(0) as usize,
828        change_writes: value
829            .get("changeWrites")
830            .and_then(Value::as_u64)
831            .unwrap_or(0) as usize,
832    })
833}
834
835fn assert_keys<'a>(
836    actual: impl Iterator<Item = &'a str>,
837    expected: impl IntoIterator<Item = &'a str>,
838) -> StorageResult<()> {
839    let mut actual = actual.collect::<Vec<_>>();
840    actual.sort();
841    let mut expected = expected.into_iter().collect::<Vec<_>>();
842    expected.sort();
843    if actual != expected {
844        return Err(StorageError::backend(format!(
845            "unexpected frame fields {}",
846            actual.join(",")
847        )));
848    }
849    Ok(())
850}
851
852fn storage_json_error(error: impl std::fmt::Display) -> StorageError {
853    StorageError::backend(error.to_string())
854}
graphrefly/adapters/agentic_memory_storage.rs

graphrefly/adapters/
agentic_memory_storage.rs
