formal_ai/
memory_sync.rs

1//! Local database sync — keep the desktop (browser IndexedDB) memory store and
2//! the CLI/native store in step without a manual export/import.
3//!
4//! Issue #347 / R5c asks that a conversation started in one surface continue in
5//! another. The surfaces already *interoperate* through the portable
6//! `formal_ai_bundle` / `demo_memory` Links-Notation files; this module adds the
7//! conflict-aware **sync** layer on top:
8//!
9//! * [`events_since`] computes the delta a puller is missing (the change-feed).
10//! * [`merge_union_by_id`] is the merge policy: events are append-only and
11//!   content-addressed, so union-by-id is conflict-free; later writes for an
12//!   existing id win the tie-break (documented, deterministic).
13//! * [`SyncStore`] is a thin file-backed store the HTTP server uses so the sync
14//!   endpoints are stateless across requests yet share one log on disk
15//!   (`FORMAL_AI_MEMORY_PATH`).
16//!
17//! Per R7 the payloads on the wire stay **Links Notation** (`demo_memory`); only
18//! the transport is REST. Nothing here introduces a non-OpenAI *external* REST
19//! surface — these are internal `/v1/memory*` sync routes.
20
21use std::path::{Path, PathBuf};
22
23use crate::memory::{export_links_notation, parse_links_notation, MemoryEvent};
24
25/// Return every event that appears strictly **after** the event `last_seen`.
26///
27/// Order is preserved. When `last_seen` is `None` (or empty), the full log is
28/// returned — a first-time puller wants everything.
29///
30/// If `last_seen` is not found in `events` (the puller saw an event this log
31/// never had — e.g. it synced from a different branch), the full log is
32/// returned so no event is silently skipped.
33#[must_use]
34pub fn events_since(events: &[MemoryEvent], last_seen: Option<&str>) -> Vec<MemoryEvent> {
35    let Some(last_seen) = last_seen.filter(|id| !id.is_empty()) else {
36        return events.to_vec();
37    };
38    events
39        .iter()
40        .position(|event| event.id == last_seen)
41        .map_or_else(|| events.to_vec(), |index| events[index + 1..].to_vec())
42}
43
44/// Merge two append-only logs by id.
45///
46/// `base` is kept in order; every event from `incoming` whose id is not already
47/// present is appended in order. Events that share an id are reconciled by
48/// [`merge_event`] (incoming non-empty fields win), so an edited event
49/// propagates without duplicating the record.
50#[must_use]
51pub fn merge_union_by_id(base: &[MemoryEvent], incoming: &[MemoryEvent]) -> Vec<MemoryEvent> {
52    let mut merged: Vec<MemoryEvent> = base.to_vec();
53    for event in incoming {
54        match merged.iter_mut().find(|existing| existing.id == event.id) {
55            Some(existing) => *existing = merge_event(existing, event),
56            None => merged.push(event.clone()),
57        }
58    }
59    merged
60}
61
62/// Tie-break for two events that share an id.
63///
64/// Keep `base` but let any non-empty field from `incoming` overwrite it. This
65/// makes "edited event" sync last-writer-wins per field while never dropping
66/// data that only one side has.
67#[must_use]
68pub fn merge_event(base: &MemoryEvent, incoming: &MemoryEvent) -> MemoryEvent {
69    fn pick(base: Option<&String>, incoming: Option<&String>) -> Option<String> {
70        match incoming {
71            Some(value) if !value.is_empty() => Some(value.clone()),
72            _ => base.cloned(),
73        }
74    }
75    let evidence = if incoming.evidence.is_empty() {
76        base.evidence.clone()
77    } else {
78        incoming.evidence.clone()
79    };
80    let payload_changed = [
81        (&base.kind, &incoming.kind),
82        (&base.role, &incoming.role),
83        (&base.intent, &incoming.intent),
84        (&base.tool, &incoming.tool),
85        (&base.inputs, &incoming.inputs),
86        (&base.outputs, &incoming.outputs),
87        (&base.content, &incoming.content),
88        (&base.sent_at, &incoming.sent_at),
89        (&base.demo_label, &incoming.demo_label),
90        (&base.conversation_id, &incoming.conversation_id),
91        (&base.conversation_title, &incoming.conversation_title),
92    ]
93    .iter()
94    .any(|(left, right)| right.as_ref().is_some_and(|value| !value.is_empty()) && left != right)
95        || (!incoming.evidence.is_empty() && base.evidence != incoming.evidence);
96    let observed_writes = base.write_count.max(1).max(incoming.write_count.max(1));
97    let write_count = if payload_changed && incoming.write_count <= base.write_count {
98        observed_writes.saturating_add(1)
99    } else {
100        observed_writes
101    };
102    MemoryEvent {
103        id: base.id.clone(),
104        kind: pick(base.kind.as_ref(), incoming.kind.as_ref()),
105        role: pick(base.role.as_ref(), incoming.role.as_ref()),
106        intent: pick(base.intent.as_ref(), incoming.intent.as_ref()),
107        tool: pick(base.tool.as_ref(), incoming.tool.as_ref()),
108        inputs: pick(base.inputs.as_ref(), incoming.inputs.as_ref()),
109        outputs: pick(base.outputs.as_ref(), incoming.outputs.as_ref()),
110        content: pick(base.content.as_ref(), incoming.content.as_ref()),
111        sent_at: pick(base.sent_at.as_ref(), incoming.sent_at.as_ref()),
112        demo_label: pick(base.demo_label.as_ref(), incoming.demo_label.as_ref()),
113        conversation_id: pick(
114            base.conversation_id.as_ref(),
115            incoming.conversation_id.as_ref(),
116        ),
117        conversation_title: pick(
118            base.conversation_title.as_ref(),
119            incoming.conversation_title.as_ref(),
120        ),
121        evidence,
122        // Access counts are monotone per event; the larger side has seen more
123        // reads, so max is the lossless merge.
124        access_count: base.access_count.max(incoming.access_count),
125        // A peer can bring a newer monotone count. Legacy/uncounted edits are
126        // recognized from their changed payload and become one durable write.
127        write_count,
128    }
129}
130
131/// Resolve the shared memory log path the server reads/writes for sync.
132///
133/// Honours `FORMAL_AI_MEMORY_PATH` and otherwise returns the platform's shared
134/// per-user memory file.
135#[must_use]
136pub fn configured_memory_path() -> Option<PathBuf> {
137    Some(crate::shared_memory::shared_memory_path())
138}
139
140/// Live chat exchanges are recorded into memory unless explicitly disabled.
141#[must_use]
142pub fn chat_recording_enabled() -> bool {
143    !matches!(
144        std::env::var("FORMAL_AI_RECORD_CHAT").as_deref(),
145        Ok("0" | "false" | "off")
146    )
147}
148
149/// A small file-backed event log used by the HTTP sync endpoints.
150///
151/// Each request loads the current log, applies its operation, and (for writes)
152/// saves it back, so the stateless server still shares one log across requests.
153#[derive(Debug, Clone, Default)]
154pub struct SyncStore {
155    path: Option<PathBuf>,
156    events: Vec<MemoryEvent>,
157}
158
159/// One client-executed tool step recovered from an agentic API transcript.
160#[derive(Debug, Clone, PartialEq, Eq)]
161pub struct RecordedToolExecution {
162    pub tool: String,
163    pub inputs: String,
164    pub outputs: String,
165}
166
167impl SyncStore {
168    /// Open the configured store, loading any existing events from disk.
169    #[must_use]
170    pub fn open() -> Self {
171        configured_memory_path().map_or_else(Self::default, |path| Self::open_at(&path))
172    }
173
174    /// Open a store at an explicit path (used by tests).
175    #[must_use]
176    pub fn open_at(path: &Path) -> Self {
177        if let Err(error) = crate::shared_memory::ensure_shared_memory_file(path) {
178            if std::env::var("FORMAL_AI_MEMORY_DEBUG").as_deref() == Ok("1") {
179                eprintln!("[memory] could not initialize {}: {error}", path.display());
180            }
181        }
182        let events = std::fs::read_to_string(path)
183            .map(|text| parse_links_notation(&text))
184            .unwrap_or_default();
185        Self {
186            path: Some(path.to_path_buf()),
187            events,
188        }
189    }
190
191    /// The events currently held.
192    #[must_use]
193    pub fn events(&self) -> &[MemoryEvent] {
194        &self.events
195    }
196
197    /// Render the log as a `demo_memory` Links-Notation document.
198    #[must_use]
199    pub fn to_links_notation(&self) -> String {
200        export_links_notation(&self.events)
201    }
202
203    /// Render only the events after `last_seen` as Links Notation (the delta a
204    /// puller applies).
205    #[must_use]
206    pub fn delta_links_notation(&self, last_seen: Option<&str>) -> String {
207        export_links_notation(&events_since(&self.events, last_seen))
208    }
209
210    /// Import a `demo_memory` document, merging by id, and persist the result.
211    /// Returns the number of events added.
212    ///
213    /// # Errors
214    /// Returns an [`std::io::Error`] when the backing file cannot be written.
215    pub fn import_links_notation(&mut self, text: &str) -> std::io::Result<usize> {
216        let incoming = parse_links_notation(text);
217        let before = self.events.len();
218        self.events = merge_union_by_id(&self.events, &incoming);
219        let added = self.events.len() - before;
220        if let Some(path) = self.path.as_deref() {
221            let mut memory =
222                crate::memory::MemoryStore::from_events(std::mem::take(&mut self.events));
223            let _ = crate::storage_policy::apply_auto_free_space_for_write(
224                &mut memory,
225                path,
226                u64::try_from(text.len()).unwrap_or(u64::MAX),
227            )?;
228            self.events = memory.events().to_vec();
229        }
230        self.persist()?;
231        Ok(added)
232    }
233
234    /// Record one live chat exchange into the shared memory log (issue #540's
235    /// live-usage loop): the user turn becomes a `message` event that
236    /// requirement learning can lift, and the assistant turn becomes a `task`
237    /// event with the exact input/output pair dreaming can replay and
238    /// generalize. Ids are stable over (prompt, answer), so retries do not
239    /// duplicate. Set `FORMAL_AI_RECORD_CHAT=0` to opt out.
240    ///
241    /// # Errors
242    /// Returns an [`std::io::Error`] when the backing file cannot be written.
243    pub fn record_chat_exchange(&mut self, prompt: &str, answer: &str) -> std::io::Result<usize> {
244        self.record_chat_exchange_with_tools(prompt, answer, &[])
245    }
246
247    /// Record a completed exchange plus the actual tool work delegated to and
248    /// returned by an agentic API client. The tool events use the same durable
249    /// schema as browser-side tool traces, and the final task cites them as
250    /// evidence. Stable ids omit transient protocol call ids, so a retried
251    /// exchange merges instead of duplicating learned evidence.
252    ///
253    /// # Errors
254    /// Returns an [`std::io::Error`] when the backing file cannot be written.
255    pub fn record_chat_exchange_with_tools(
256        &mut self,
257        prompt: &str,
258        answer: &str,
259        tools: &[RecordedToolExecution],
260    ) -> std::io::Result<usize> {
261        if self.path.is_none() || !chat_recording_enabled() {
262            return Ok(0);
263        }
264        let seed = format!("{prompt}\0{answer}");
265        let user_id = crate::engine::stable_id("chat_user", &seed);
266        let mut recorded = vec![MemoryEvent {
267            id: user_id.clone(),
268            kind: Some(String::from("message")),
269            role: Some(String::from("user")),
270            content: Some(prompt.to_owned()),
271            write_count: 1,
272            ..MemoryEvent::default()
273        }];
274        let mut evidence = vec![user_id.clone()];
275        for execution in tools {
276            let tool_seed = format!(
277                "{prompt}\0{}\0{}\0{}",
278                execution.tool, execution.inputs, execution.outputs
279            );
280            let id = crate::engine::stable_id("chat_tool", &tool_seed);
281            evidence.push(id.clone());
282            recorded.push(MemoryEvent {
283                id,
284                kind: Some(String::from("tool_call")),
285                role: Some(String::from("assistant")),
286                intent: Some(String::from("execute_tool")),
287                tool: Some(execution.tool.clone()),
288                inputs: Some(execution.inputs.clone()),
289                outputs: Some(execution.outputs.clone()),
290                content: Some(format!("tool:{}", execution.tool)),
291                evidence: vec![user_id.clone()],
292                write_count: 1,
293                ..MemoryEvent::default()
294            });
295        }
296        recorded.push(MemoryEvent {
297            id: crate::engine::stable_id("chat_task", &seed),
298            kind: Some(String::from("task")),
299            role: Some(String::from("assistant")),
300            intent: Some(String::from("solve")),
301            inputs: Some(prompt.to_owned()),
302            outputs: Some(answer.to_owned()),
303            evidence,
304            write_count: 1,
305            ..MemoryEvent::default()
306        });
307        let before = self.events.len();
308        self.events = merge_union_by_id(&self.events, &recorded);
309        let added = self.events.len() - before;
310        if added > 0 {
311            self.persist()?;
312        }
313        Ok(added)
314    }
315
316    fn persist(&self) -> std::io::Result<()> {
317        let Some(path) = self.path.as_ref() else {
318            return Ok(());
319        };
320        // Locked atomic write (issue #540 §6): the HTTP handlers and the
321        // background dreaming thread share this log.
322        crate::memory::write_locked_atomic(path, &self.to_links_notation())
323    }
324}
formal_ai/memory_sync.rs

formal_ai/
memory_sync.rs
